Commit 48a29c72 authored by Dominik Charousset's avatar Dominik Charousset

maintenance & documentation update

this patch renames `message_id_t` to `message_id`,
condenses the `receive` functions, and updates the documentation
parent ef0b744f
......@@ -137,7 +137,6 @@ set(LIBCPPA_SRC
src/primitive_variant.cpp
src/process_information.cpp
src/protocol.cpp
src/receive.cpp
src/recursive_queue_node.cpp
src/ref_counted.cpp
src/response_handle.cpp
......
......@@ -80,7 +80,6 @@ cppa/group.hpp
cppa/guard_expr.hpp
cppa/intrusive/blocking_single_reader_queue.hpp
cppa/intrusive/single_reader_queue.hpp
cppa/intrusive_fwd_ptr.hpp
cppa/intrusive_ptr.hpp
cppa/local_actor.hpp
cppa/logging.hpp
......@@ -244,7 +243,6 @@ src/partial_function.cpp
src/primitive_variant.cpp
src/process_information.cpp
src/protocol.cpp
src/receive.cpp
src/recursive_queue_node.cpp
src/ref_counted.cpp
src/response_handle.cpp
......
......@@ -86,7 +86,7 @@ class actor : public channel {
* @pre <tt>id.valid()</tt>
*/
virtual void sync_enqueue(const actor_ptr& sender,
message_id_t id,
message_id id,
any_tuple msg) = 0;
/**
......@@ -96,7 +96,7 @@ class actor : public channel {
* its state to @p pending in response to the enqueue operation.
*/
virtual bool chained_sync_enqueue(const actor_ptr& sender,
message_id_t id,
message_id id,
any_tuple msg);
/**
......
......@@ -135,7 +135,7 @@ class actor_companion_mixin : public Base {
}
void sync_enqueue(const actor_ptr& sender,
message_id_t id,
message_id id,
any_tuple msg) {
util::shared_lock_guard<lock_type> guard(m_lock);
if (m_parent) {
......@@ -151,15 +151,15 @@ class actor_companion_mixin : public Base {
"management instead!");
}
void dequeue(behavior&) { throw_no_recv(); }
void dequeue(partial_function&) { throw_no_recv(); }
void dequeue(behavior&) {
throw_no_recv();
}
void dequeue_response(behavior&, message_id_t) {
void dequeue_response(behavior&, message_id) {
throw_no_recv();
}
void become_waiting_for(behavior&&, message_id_t) {
void become_waiting_for(behavior, message_id) {
throw_no_become();
}
......
......@@ -56,10 +56,18 @@ class behavior {
public:
/** @cond PRIVATE */
typedef intrusive_ptr<detail::behavior_impl> impl_ptr;
inline auto as_behavior_impl() const -> const impl_ptr&;
inline behavior(impl_ptr ptr);
static constexpr bool may_have_timeout = true;
/** @endcond */
behavior() = default;
behavior(behavior&&) = default;
behavior(const behavior&) = default;
......@@ -68,49 +76,39 @@ class behavior {
behavior(const partial_function& fun);
inline behavior(impl_ptr ptr) : m_impl(std::move(ptr)) { }
template<typename F>
behavior(const timeout_definition<F>& arg)
: m_impl(detail::new_default_behavior_impl(detail::dummy_match_expr{},
arg.timeout,
arg.handler) ) { }
behavior(const timeout_definition<F>& arg);
template<typename F>
behavior(util::duration d, F f)
: m_impl(detail::new_default_behavior_impl(detail::dummy_match_expr{}, d, f)) { }
template<typename... Cases>
behavior(const match_expr<Cases...>& arg0)
: m_impl(detail::new_default_behavior_impl(arg0, util::duration(), []() { })) { }
behavior(util::duration d, F f);
template<typename... Cases, typename Arg1, typename... Args>
behavior(const match_expr<Cases...>& arg0, const Arg1& arg1, const Args&... args)
: m_impl(detail::match_expr_concat(arg0, arg1, args...)) { }
template<typename... Cases, typename... Ts>
behavior(const match_expr<Cases...>& arg, const Ts&... args);
inline void handle_timeout() {
m_impl->handle_timeout();
}
inline const util::duration& timeout() const {
return m_impl->timeout();
}
/**
* @brief Invokes the timeout callback.
*/
inline void handle_timeout();
inline bool undefined() const {
return m_impl == nullptr || m_impl->timeout().valid() == false;
}
/**
* @brief Returns the duration after which receives using
* this behavior should time out.
*/
inline const util::duration& timeout() const;
/**
* @copydoc partial_function::operator()()
*/
template<typename T>
inline bool operator()(T&& arg) {
return (m_impl) && m_impl->invoke(std::forward<T>(arg));
}
inline bool operator()(T&& arg);
/**
* @brief Adds a continuation to this behavior that is executed
* whenever this behavior was successfully applied to
* a message.
*/
template<typename F>
inline behavior add_continuation(F fun) {
return {new detail::continuation_decorator<F>(std::move(fun), m_impl)};
}
const impl_ptr& as_behavior_impl() const { return m_impl; }
inline behavior add_continuation(F fun);
private:
......@@ -118,12 +116,60 @@ class behavior {
};
template<typename... Lhs, typename F>
inline behavior operator,(const match_expr<Lhs...>& lhs,
/**
* @brief Creates a behavior from a match expression and a timeout definition.
* @relates behavior
*/
template<typename... Cases, typename F>
inline behavior operator,(const match_expr<Cases...>& lhs,
const timeout_definition<F>& rhs) {
return match_expr_convert(lhs, rhs);
}
/******************************************************************************
* inline and template member function implementations *
******************************************************************************/
template<typename F>
behavior::behavior(const timeout_definition<F>& arg)
: m_impl(detail::new_default_behavior_impl(detail::dummy_match_expr{},
arg.timeout,
arg.handler) ) { }
template<typename F>
behavior::behavior(util::duration d, F f)
: m_impl(detail::new_default_behavior_impl(detail::dummy_match_expr{}, d, f)) { }
template<typename... Cases, typename... Ts>
behavior::behavior(const match_expr<Cases...>& arg, const Ts&... args)
: m_impl(detail::match_expr_concat(arg, args...)) { }
inline behavior::behavior(impl_ptr ptr) : m_impl(std::move(ptr)) { }
inline void behavior::handle_timeout() {
m_impl->handle_timeout();
}
inline const util::duration& behavior::timeout() const {
return m_impl->timeout();
}
template<typename T>
inline bool behavior::operator()(T&& arg) {
return (m_impl) && m_impl->invoke(std::forward<T>(arg));
}
inline auto behavior::as_behavior_impl() const -> const impl_ptr& {
return m_impl;
}
template<typename F>
inline behavior behavior::add_continuation(F fun) {
return {new detail::continuation_decorator<F>(std::move(fun), m_impl)};
}
} // namespace cppa
#endif // CPPA_BEHAVIOR_HPP
......@@ -67,12 +67,6 @@ class channel : public ref_counted {
virtual ~channel();
private:
// channel is a sealed class and the only
// allowed subclasses are actor and group.
channel() = default;
};
/**
......@@ -84,8 +78,8 @@ typedef intrusive_ptr<channel> channel_ptr;
/**
* @brief Convenience alias.
*/
template<typename T>
using enable_if_channel = std::enable_if<std::is_base_of<channel,T>::value>;
template<typename T, typename R = void>
using enable_if_channel = std::enable_if<std::is_base_of<channel,T>::value,R>;
} // namespace cppa
......
......@@ -191,11 +191,20 @@
*
* @defgroup MessageHandling Message handling.
*
* This is the beating heart of @p libcppa. Actor programming is all about
* message handling.
* @brief This is the beating heart of @p libcppa. Actor programming is
* all about message handling.
*
* A message in @p libcppa is a n-tuple of values (with size >= 1). You can use
* almost every type in a messages.
* almost every type in a messages - as long as it is announced, i.e., known
* by libcppa's type system.
*
* @defgroup BlockingAPI Blocking API.
*
* @brief Blocking functions to receive messages.
*
* The blocking API of libcppa is intended to be used for migrating
* previously threaded applications. When writing new code, you should use
* ibcppas nonblocking become/unbecome API.
*
* @section Send Send messages
*
......@@ -414,15 +423,14 @@ namespace cppa {
namespace detail {
template<typename T>
inline void send_impl(T* whom, any_tuple&& what) {
if (whom) self->send_message(whom, std::move(what));
inline void send_impl(T* ptr, any_tuple&& arg) {
if (ptr) self->send_message(ptr, std::move(arg));
}
template<typename T, typename... Ts>
inline void send_tpl_impl(T* whom, Ts&&... what) {
inline void send_tpl_impl(T* ptr, Ts&&... args) {
static_assert(sizeof...(Ts) > 0, "no message to send");
if (whom) self->send_message(whom,
make_any_tuple(std::forward<Ts>(what)...));
if (ptr) self->send_message(ptr, make_any_tuple(std::forward<Ts>(args)...));
}
} // namespace detail
......@@ -438,7 +446,7 @@ inline void send_tpl_impl(T* whom, Ts&&... what) {
* @param what Message content as tuple.
*/
template<class C, typename... Ts>
inline typename std::enable_if<std::is_base_of<channel, C>::value>::type
inline typename enable_if_channel<C>::type
send_tuple(const intrusive_ptr<C>& whom, any_tuple what) {
detail::send_impl(whom.get(), std::move(what));
}
......@@ -450,9 +458,9 @@ send_tuple(const intrusive_ptr<C>& whom, any_tuple what) {
* @pre <tt>sizeof...(Ts) > 0</tt>
*/
template<class C, typename... Ts>
inline typename std::enable_if<std::is_base_of<channel, C>::value>::type
send(const intrusive_ptr<C>& whom, Ts&&... what) {
detail::send_tpl_impl(whom.get(), std::forward<Ts>(what)...);
inline typename enable_if_channel<C>::type
send(const intrusive_ptr<C>& whom, Ts&&... args) {
detail::send_tpl_impl(whom.get(), std::forward<Ts>(args)...);
}
/**
......@@ -464,7 +472,7 @@ send(const intrusive_ptr<C>& whom, Ts&&... what) {
* @pre <tt>sizeof...(Ts) > 0</tt>
*/
template<class C, typename... Ts>
inline typename std::enable_if<std::is_base_of<channel, C>::value>::type
inline typename enable_if_channel<C>::type
send_tuple_as(const actor_ptr& from, const intrusive_ptr<C>& whom, any_tuple what) {
if (whom) whom->enqueue(from.get(), std::move(what));
}
......@@ -478,7 +486,7 @@ send_tuple_as(const actor_ptr& from, const intrusive_ptr<C>& whom, any_tuple wha
* @pre <tt>sizeof...(Ts) > 0</tt>
*/
template<class C, typename... Ts>
inline typename std::enable_if<std::is_base_of<channel, C>::value>::type
inline typename enable_if_channel<C>::type
send_as(const actor_ptr& from, const intrusive_ptr<C>& whom, Ts&&... what) {
send_tuple_as(from, whom, make_any_tuple(std::forward<Ts>(what)...));
}
......@@ -495,8 +503,7 @@ send_as(const actor_ptr& from, const intrusive_ptr<C>& whom, Ts&&... what) {
* @returns @p whom.
*/
template<class C>
inline typename std::enable_if<std::is_base_of<channel, C>::value,
const intrusive_ptr<C>& >::type
inline typename enable_if_channel<C,const intrusive_ptr<C>&>::type
operator<<(const intrusive_ptr<C>& whom, any_tuple what) {
send_tuple(whom, std::move(what));
return whom;
......@@ -683,18 +690,20 @@ inline void delayed_reply(const std::chrono::duration<Rep, Period>& rtime,
delayed_reply_tuple(rtime, make_any_tuple(std::forward<Ts>(what)...));
}
/** @} */
/**
* @}
*/
// matches "send(this, ...)" and "send(self, ...)"
inline void send_tuple(local_actor* whom, any_tuple what) {
inline void send_tuple(channel* whom, any_tuple what) {
detail::send_impl(whom, std::move(what));
}
template<typename... Ts>
inline void send(local_actor* whom, Ts&&... args) {
inline void send(channel* whom, Ts&&... args) {
detail::send_tpl_impl(whom, std::forward<Ts>(args)...);
}
inline const self_type& operator<<(const self_type& s, any_tuple what) {
detail::send_impl(s.get(), std::move(what));
detail::send_impl(static_cast<channel*>(s.get()), std::move(what));
return s;
}
inline actor_ptr eval_sopts(spawn_options opts, actor_ptr ptr) {
......@@ -844,15 +853,24 @@ void shutdown(); // note: implemented in singleton_manager.cpp
*/
inline void quit_actor(const actor_ptr& whom, std::uint32_t reason) {
CPPA_REQUIRE(reason != exit_reason::normal);
send(whom, atom("EXIT"), reason);
send(whom.get(), atom("EXIT"), reason);
}
/**
* @brief Sets the actor's behavior and discards the previous behavior
* unless {@link keep_behavior} is given as first argument.
*/
template<typename... Ts>
inline void become(Ts&&... args) {
self->become(std::forward<Ts>(args)...);
}
inline void unbecome() { self->unbecome(); }
/**
* @brief Returns to a previous behavior if available.
*/
inline void unbecome() {
self->unbecome();
}
struct actor_ostream {
......@@ -883,9 +901,11 @@ inline const actor_ostream& operator<<(const actor_ostream& o, const any_tuple&
}
template<typename T>
inline typename std::enable_if< !std::is_convertible<T,std::string>::value
inline typename std::enable_if<
!std::is_convertible<T,std::string>::value
&& !std::is_convertible<T,any_tuple>::value,
const actor_ostream&>::type
const actor_ostream&
>::type
operator<<(const actor_ostream& o, T&& arg) {
return o.write(std::to_string(std::forward<T>(arg)));
}
......
......@@ -54,14 +54,14 @@ class behavior_stack
behavior_stack(const behavior_stack&) = delete;
behavior_stack& operator=(const behavior_stack&) = delete;
typedef std::pair<behavior,message_id_t> element_type;
typedef std::pair<behavior,message_id> element_type;
public:
behavior_stack() = default;
// @pre expected_response.valid()
option<behavior&> sync_handler(message_id_t expected_response);
option<behavior&> sync_handler(message_id expected_response);
// erases the last asynchronous message handler
void pop_async_back();
......@@ -69,7 +69,7 @@ class behavior_stack
void clear();
// erases the synchronous response handler associated with @p rid
void erase(message_id_t rid) {
void erase(message_id rid) {
erase_if([=](const element_type& e) { return e.second == rid; });
}
......@@ -81,7 +81,7 @@ class behavior_stack
}
inline void push_back(behavior&& what,
message_id_t response_id = message_id_t::invalid) {
message_id response_id = message_id::invalid) {
m_elements.emplace_back(std::move(what), response_id);
}
......
......@@ -51,24 +51,15 @@ struct receive_while_helper {
template<typename S>
receive_while_helper(S&& stmt) : m_stmt(std::forward<S>(stmt)) { }
void operator()(behavior& bhvr) {
template<typename... Ts>
void operator()(Ts&&... args) {
static_assert(sizeof...(Ts) > 0,
"operator() requires at least one argument");
behavior tmp = match_expr_convert(std::forward<Ts>(args)...);
local_actor* sptr = self;
while (m_stmt()) sptr->dequeue(bhvr);
while (m_stmt()) sptr->dequeue(tmp);
}
void operator()(partial_function& fun) {
local_actor* sptr = self;
while (m_stmt()) sptr->dequeue(fun);
}
template<typename Arg0, typename... Args>
void operator()(Arg0&& arg0, Args&&... args) {
auto tmp = match_expr_convert(std::forward<Arg0>(arg0),
std::forward<Args>(args)...);
(*this)(tmp);
}
};
template<typename T>
......@@ -81,21 +72,11 @@ class receive_for_helper {
receive_for_helper(T& first, const T& last) : begin(first), end(last) { }
void operator()(behavior& bhvr) {
template<typename... Ts>
void operator()(Ts&&... args) {
auto tmp = match_expr_convert(std::forward<Ts>(args)...);
local_actor* sptr = self;
for ( ; begin != end; ++begin) sptr->dequeue(bhvr);
}
void operator()(partial_function& fun) {
local_actor* sptr = self;
for ( ; begin != end; ++begin) sptr->dequeue(fun);
}
template<typename Arg0, typename... Args>
void operator()(Arg0&& arg0, Args&&... args) {
auto tmp = match_expr_convert(std::forward<Arg0>(arg0),
std::forward<Args>(args)...);
(*this)(tmp);
for ( ; begin != end; ++begin) sptr->dequeue(tmp);
}
};
......@@ -106,10 +87,7 @@ class do_receive_helper {
public:
template<typename... Args>
do_receive_helper(Args&&... args)
: m_bhvr(match_expr_convert(std::forward<Args>(args)...)) {
}
inline do_receive_helper(behavior&& bhvr) : m_bhvr(std::move(bhvr)) { }
do_receive_helper(do_receive_helper&&) = default;
......
......@@ -76,7 +76,7 @@ class receive_policy {
template<class Client, class Fun>
bool invoke_from_cache(Client* client,
Fun& fun,
message_id_t awaited_response = message_id_t()) {
message_id awaited_response = message_id{}) {
std::integral_constant<receive_policy_flag,Client::receive_flag> policy;
auto i = m_cache.begin();
auto e = m_cache.end();
......@@ -108,7 +108,7 @@ class receive_policy {
bool invoke(Client* client,
pointer node_ptr,
Fun& fun,
message_id_t awaited_response = message_id_t()) {
message_id awaited_response = message_id()) {
smart_pointer node(node_ptr);
std::integral_constant<receive_policy_flag,Client::receive_flag> policy;
switch (this->handle_message(client, node.get(), fun,
......@@ -176,7 +176,7 @@ class receive_policy {
}
template<class Client>
void receive(Client* client, behavior& bhvr, message_id_t mid) {
void receive(Client* client, behavior& bhvr, message_id mid) {
CPPA_REQUIRE(mid.is_response());
if (!invoke_from_cache(client, bhvr, mid)) {
if (bhvr.timeout().valid()) {
......@@ -231,7 +231,7 @@ class receive_policy {
template<class Client>
filter_result filter_msg(Client* client, pointer node) {
const any_tuple& msg = node->msg;
auto message_id = node->mid;
auto mid = node->mid;
auto& arr = detail::static_types_array<atom_value,std::uint32_t>::arr;
if ( msg.size() == 2
&& msg.type_at(0) == arr[0]
......@@ -239,7 +239,7 @@ class receive_policy {
auto v0 = msg.get_as<atom_value>(0);
auto v1 = msg.get_as<std::uint32_t>(1);
if (v0 == atom("EXIT")) {
CPPA_REQUIRE(!message_id.valid());
CPPA_REQUIRE(!mid.valid());
if (client->m_trap_exit == false) {
if (v1 != exit_reason::normal) {
client->quit(v1);
......@@ -249,7 +249,7 @@ class receive_policy {
}
}
else if (v0 == atom("SYNC_TOUT")) {
CPPA_REQUIRE(!message_id.valid());
CPPA_REQUIRE(!mid.valid());
return client->waits_for_timeout(v1) ? timeout_message
: expired_timeout_message;
}
......@@ -257,11 +257,11 @@ class receive_policy {
else if ( msg.size() == 1
&& msg.type_at(0) == arr[0]
&& msg.get_as<atom_value>(0) == atom("TIMEOUT")
&& message_id.is_response()) {
&& mid.is_response()) {
return timeout_response_message;
}
if (message_id.is_response()) {
return (client->awaits(message_id)) ? sync_response
if (mid.is_response()) {
return (client->awaits(mid)) ? sync_response
: expired_sync_response;
}
return ordinary_message;
......@@ -340,7 +340,7 @@ class receive_policy {
handle_message_result handle_message(Client* client,
pointer node,
Fun& fun,
message_id_t awaited_response,
message_id awaited_response,
Policy policy ) {
bool handle_sync_failure_on_mismatch = true;
if (hm_should_skip(node, policy)) {
......
......@@ -57,7 +57,7 @@ class recursive_queue_node : public memory_cached<memory_managed,recursive_queue
bool marked; // denotes if this node is currently processed
actor_ptr sender; // points to the sender of msg
any_tuple msg; // 'content field'
message_id_t mid;
message_id mid;
recursive_queue_node(recursive_queue_node&&) = delete;
recursive_queue_node(const recursive_queue_node&) = delete;
......@@ -75,7 +75,7 @@ class recursive_queue_node : public memory_cached<memory_managed,recursive_queue
recursive_queue_node(const actor_ptr& sptr,
any_tuple data,
message_id_t id = message_id_t{});
message_id id = message_id{});
};
......
......@@ -38,14 +38,13 @@ namespace cppa { namespace detail {
struct scheduled_actor_dummy : scheduled_actor {
scheduled_actor_dummy();
void enqueue(const actor_ptr&, any_tuple);
void sync_enqueue(const actor_ptr&, message_id_t, any_tuple);
void sync_enqueue(const actor_ptr&, message_id, any_tuple);
resume_result resume(util::fiber*, actor_ptr&);
void quit(std::uint32_t);
void dequeue(behavior&);
void dequeue(partial_function&);
void dequeue_response(behavior&, message_id_t);
void dequeue_response(behavior&, message_id);
void do_become(behavior&&, bool);
void become_waiting_for(behavior&&, message_id_t);
void become_waiting_for(behavior, message_id);
bool has_behavior();
scheduled_actor_type impl_type();
};
......
......@@ -45,7 +45,7 @@ struct sync_request_bouncer {
std::uint32_t rsn;
constexpr sync_request_bouncer(std::uint32_t r)
: rsn(r == exit_reason::not_exited ? exit_reason::normal : r) { }
inline void operator()(const actor_ptr& sender, const message_id_t& mid) const {
inline void operator()(const actor_ptr& sender, const message_id& mid) const {
CPPA_REQUIRE(rsn != exit_reason::not_exited);
if (mid.is_request() && sender != nullptr) {
actor_ptr nobody;
......
......@@ -65,12 +65,7 @@ class event_based_actor : public scheduled_actor {
/**
* @copydoc dequeue(behavior&)
*/
void dequeue(partial_function&); //override
/**
* @copydoc dequeue(behavior&)
*/
void dequeue_response(behavior&, message_id_t);
void dequeue_response(behavior&, message_id);
resume_result resume(util::fiber*, actor_ptr&); //override
......@@ -130,7 +125,7 @@ class event_based_actor : public scheduled_actor {
void do_become(behavior&& bhvr, bool discard_old);
void become_waiting_for(behavior&& bhvr, message_id_t mf);
void become_waiting_for(behavior bhvr, message_id mf);
private:
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation, either version 3 of the License *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#ifndef CPPA_INTRUSIVE_FWD_PTR_HPP
#define CPPA_INTRUSIVE_FWD_PTR_HPP
#include <memory>
#include "cppa/util/comparable.hpp"
namespace cppa {
/**
* @brief A reference counting smart pointer implementation that
* can be used with forward declarated types.
* @warning libcppa assumes T is derived from ref_counted and
* implicitly converts intrusive_fwd_ptr to intrusive_ptr
* @relates intrusive_ptr
* @relates ref_counted
*/
template<typename T, typename Ref, typename Deref>
class intrusive_fwd_ptr : util::comparable<intrusive_fwd_ptr<T,Ref,Deref> >,
util::comparable<intrusive_fwd_ptr<T,Ref,Deref>, const T*>,
util::comparable<intrusive_fwd_ptr<T,Ref,Deref>, std::nullptr_t> {
public:
typedef T* pointer;
typedef const T* const_pointer;
typedef T element_type;
typedef T& reference;
typedef const T& const_reference;
intrusive_fwd_ptr(pointer raw_ptr = nullptr, Ref ref = Ref(), Deref deref = Deref())
: m_ref(std::move(ref)), m_deref(std::move(deref)) { set_ptr(raw_ptr); }
intrusive_fwd_ptr(intrusive_fwd_ptr&& other)
: m_ref(std::move(other.m_ref))
, m_deref(std::move(m_deref))
, m_ptr(other.release()) { }
intrusive_fwd_ptr(const intrusive_fwd_ptr& other)
: m_ref(other.m_ref), m_deref(other.m_deref) { set_ptr(other.get()); }
~intrusive_fwd_ptr() { if (m_ptr) m_deref(m_ptr); }
inline void swap(intrusive_fwd_ptr& other) {
std::swap(m_ptr, other.m_ptr);
}
/**
* @brief Returns the raw pointer without modifying reference count.
*/
pointer release() {
auto result = m_ptr;
m_ptr = nullptr;
return result;
}
/**
* @brief Sets this pointer to @p ptr without modifying reference count.
*/
void adopt(pointer ptr) {
reset();
m_ptr = ptr;
}
void reset(pointer new_value = nullptr) {
if (m_ptr) m_deref(m_ptr);
set_ptr(new_value);
}
template<typename... Args>
void emplace(Args&&... args) {
reset(new T(std::forward<Args>(args)...));
}
intrusive_fwd_ptr& operator=(pointer ptr) {
reset(ptr);
return *this;
}
intrusive_fwd_ptr& operator=(intrusive_fwd_ptr&& other) {
swap(other);
return *this;
}
intrusive_fwd_ptr& operator=(const intrusive_fwd_ptr& other) {
intrusive_fwd_ptr tmp(other);
swap(tmp);
return *this;
}
inline pointer get() const { return m_ptr; }
inline pointer operator->() const { return m_ptr; }
inline reference operator*() const { return *m_ptr; }
inline bool operator!() const { return m_ptr == nullptr; }
inline explicit operator bool() const { return m_ptr != nullptr; }
inline ptrdiff_t compare(const_pointer ptr) const {
return static_cast<ptrdiff_t>(get() - ptr);
}
inline ptrdiff_t compare(const intrusive_fwd_ptr& other) const {
return compare(other.get());
}
inline ptrdiff_t compare(std::nullptr_t) const {
return reinterpret_cast<ptrdiff_t>(get());
}
private:
Ref m_ref;
Deref m_deref;
pointer m_ptr;
inline void set_ptr(pointer raw_ptr) {
m_ptr = raw_ptr;
if (raw_ptr) m_ref(raw_ptr);
}
};
/**
* @relates intrusive_ptr
*/
template<typename T, typename Ref, typename Deref>
inline bool operator==(const intrusive_fwd_ptr<T,Ref,Deref>& lhs,
const intrusive_fwd_ptr<T,Ref,Deref>& rhs) {
return lhs.get() == rhs.get();
}
/**
* @relates intrusive_ptr
*/
template<typename T, typename Ref, typename Deref>
inline bool operator!=(const intrusive_fwd_ptr<T,Ref,Deref>& lhs,
const intrusive_fwd_ptr<T,Ref,Deref>& rhs) {
return !(lhs == rhs);
}
} // namespace cppa
#endif // CPPA_INTRUSIVE_FWD_PTR_HPP
......@@ -36,7 +36,6 @@
#include <stdexcept>
#include <type_traits>
#include "cppa/intrusive_fwd_ptr.hpp"
#include "cppa/util/comparable.hpp"
namespace cppa {
......@@ -73,12 +72,6 @@ class intrusive_ptr : util::comparable<intrusive_ptr<T> >,
intrusive_ptr(const intrusive_ptr& other) { set_ptr(other.get()); }
template<typename Y, typename Ref, typename Deref>
intrusive_ptr(intrusive_fwd_ptr<Y,Ref,Deref> other) : m_ptr(other.release()) {
static_assert(std::is_convertible<Y*, T*>::value,
"Y* is not assignable to T*");
}
template<typename Y>
intrusive_ptr(intrusive_ptr<Y> other) : m_ptr(other.release()) {
static_assert(std::is_convertible<Y*, T*>::value,
......
......@@ -42,11 +42,11 @@
#include "cppa/message_id.hpp"
#include "cppa/match_expr.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/memory_cached.hpp"
#include "cppa/response_handle.hpp"
#include "cppa/partial_function.hpp"
#include "cppa/util/duration.hpp"
#include "cppa/memory_cached.hpp"
#include "cppa/detail/behavior_stack.hpp"
#include "cppa/detail/recursive_queue_node.hpp"
......@@ -89,11 +89,6 @@ class local_actor : public extend<actor,local_actor>::with<memory_cached> {
typedef combined_type super;
friend class scheduler;
friend class detail::memory;
friend class detail::receive_policy;
template<typename> friend class detail::basic_memory_cache;
public:
/**
......@@ -119,36 +114,11 @@ class local_actor : public extend<actor,local_actor>::with<memory_cached> {
* linked actors, sets its state to @c exited and finishes execution.
* @param reason Exit reason that will be send to
* linked actors and monitors.
* @note Throws {@link actor_exited} to unwind the stack
* when called in context-switching or thread-based actors.
*/
virtual void quit(std::uint32_t reason = exit_reason::normal) = 0;
/**
* @brief Removes the first element from the mailbox @p pfun is defined
* for and invokes @p pfun with the removed element.
* Blocks until a matching message arrives if @p pfun is not
* defined for any message in the actor's mailbox.
* @param pfun A partial function denoting the actor's response to the
* next incoming message.
* @warning You should not call this member function by hand.
* Use the {@link cppa::receive receive} function or
* the @p become member function in case of event-based actors.
*/
virtual void dequeue(partial_function& pfun) = 0;
/**
* @brief Removes the first element from the mailbox @p bhvr is defined
* for and invokes @p bhvr with the removed element.
* Blocks until either a matching message arrives if @p bhvr is not
* defined for any message in the actor's mailbox or until a
* timeout occurs.
* @param bhvr A partial function with optional timeout denoting the
* actor's response to the next incoming message.
* @warning You should not call this member function by hand.
* Use the {@link cppa::receive receive} function or
* the @p become member function in case of event-based actors.
*/
virtual void dequeue(behavior& bhvr) = 0;
/**
* @brief Checks whether this actor traps exit messages.
*/
......@@ -191,47 +161,13 @@ class local_actor : public extend<actor,local_actor>::with<memory_cached> {
* @param whom The actor that should be monitored by this actor.
* @note Each call to @p monitor creates a new, independent monitor.
*/
void monitor(actor_ptr whom);
void monitor(const actor_ptr& whom);
/**
* @brief Removes a monitor from @p whom.
* @param whom A monitored actor.
*/
void demonitor(actor_ptr whom);
// become/unbecome API
/**
* @brief Sets the actor's behavior to @p bhvr and discards the
* previous behavior if the policy is {@link discard_behavior}.
*/
template<bool Discard>
inline void become(behavior_policy<Discard>, behavior bhvr);
/**
* @brief Sets the actor's behavior and discards the
* previous behavior if the policy is {@link discard_behavior}.
*/
template<bool Discard, typename T0, typename T1, typename... Ts>
inline void become(behavior_policy<Discard>, T0&& a0, T1&& a1, Ts&&... as);
/**
* @brief Sets the actor's behavior;
* equal to <tt>become(discard_old, bhvr</tt>.
*/
inline void become(behavior bhvr);
/**
* @brief Sets the actor's behavior;
* equal to <tt>become(discard_old, bhvr</tt>.
*/
template<typename... Cases, typename... Ts>
inline void become(match_expr<Cases...> a, Ts&&... as);
/**
* @brief Returns to a previous behavior if available.
*/
inline void unbecome();
void demonitor(const actor_ptr& whom);
/**
* @brief Can be overridden to initialize an actor before any
......@@ -256,6 +192,7 @@ class local_actor : public extend<actor,local_actor>::with<memory_cached> {
std::vector<group_ptr> joined_groups();
/**
* @ingroup BlockingAPI
* @brief Executes an actor's behavior stack until it is empty.
*
* This member function allows thread-based and context-switching actors
......@@ -308,146 +245,108 @@ class local_actor : public extend<actor,local_actor>::with<memory_cached> {
else quit(exit_reason::unhandled_sync_failure);
}
// library-internal members and member functions that shall
// not appear in the documentation
/** @cond PRIVATE */
virtual void dequeue(behavior& bhvr) = 0;
inline void dequeue(behavior&& bhvr);
virtual void dequeue_response(behavior&, message_id) = 0;
inline void dequeue_response(behavior&&, message_id);
# ifndef CPPA_DOCUMENTATION
template<bool Discard, typename... Ts>
void become(behavior_policy<Discard>, Ts&&... args);
// backward compatiblity, handle_response was part of local_actor
// until version 0.6
sync_handle_helper handle_response(const message_future& mf);
template<typename T, typename... Ts>
void become(T arg, Ts&&... args);
inline void unbecome();
local_actor(bool is_scheduled = false);
virtual bool initialized() const = 0;
inline bool chaining_enabled() {
return m_chaining && !m_chained_actor;
}
inline bool chaining_enabled();
inline void send_message(channel* whom, any_tuple&& what) {
whom->enqueue(this, std::move(what));
}
inline void send_message(channel* whom, any_tuple&& what);
inline void send_message(const actor_ptr& whom, any_tuple&& what) {
if (chaining_enabled()) {
if (whom->chained_enqueue(this, std::move(what))) {
m_chained_actor = whom;
}
}
else whom->enqueue(this, std::move(what));
}
inline void send_message(actor* whom, any_tuple&& what);
inline message_id_t send_sync_message(const actor_ptr& whom, any_tuple&& what) {
auto id = ++m_last_request_id;
CPPA_REQUIRE(id.is_request());
if (chaining_enabled()) {
if (whom->chained_sync_enqueue(this, id, std::move(what))) {
m_chained_actor = whom;
}
}
else whom->sync_enqueue(this, id, std::move(what));
auto awaited_response = id.response_id();
m_pending_responses.push_back(awaited_response);
return awaited_response;
}
inline message_id send_sync_message(const actor_ptr& whom,
any_tuple&& what);
message_id_t send_timed_sync_message(const actor_ptr& whom,
message_id send_timed_sync_message(const actor_ptr& whom,
const util::duration& rel_time,
any_tuple&& what);
// returns 0 if last_dequeued() is an asynchronous or sync request message,
// a response id generated from the request id otherwise
inline message_id_t get_response_id() {
auto id = m_current_node->mid;
return (id.is_request()) ? id.response_id() : message_id_t();
}
inline message_id get_response_id();
void reply_message(any_tuple&& what);
void forward_message(const actor_ptr& new_receiver);
inline const actor_ptr& chained_actor() {
return m_chained_actor;
}
inline const actor_ptr& chained_actor();
inline void chained_actor(const actor_ptr& value) {
m_chained_actor = value;
}
inline void chained_actor(const actor_ptr& value);
inline bool awaits(message_id_t response_id) {
CPPA_REQUIRE(response_id.is_response());
return std::any_of(m_pending_responses.begin(),
m_pending_responses.end(),
[=](message_id_t other) {
return response_id == other;
});
}
inline bool awaits(message_id response_id);
inline void mark_arrived(message_id_t response_id) {
auto last = m_pending_responses.end();
auto i = std::find(m_pending_responses.begin(), last, response_id);
if (i != last) m_pending_responses.erase(i);
}
inline void mark_arrived(message_id response_id);
virtual void dequeue_response(behavior&, message_id_t) = 0;
virtual void become_waiting_for(behavior, message_id) = 0;
inline void dequeue_response(behavior&& bhvr, message_id_t mid) {
behavior tmp{std::move(bhvr)};
dequeue_response(tmp, mid);
}
inline detail::behavior_stack& bhvr_stack();
virtual void become_waiting_for(behavior&&, message_id_t) = 0;
protected:
inline detail::behavior_stack& bhvr_stack() {
return m_bhvr_stack;
}
virtual void do_become(behavior&& bhvr, bool discard_old) = 0;
protected:
inline void do_become(const behavior& bhvr, bool discard_old);
inline void remove_handler(message_id id);
void cleanup(std::uint32_t reason);
// true if this actor uses the chained_send optimization
bool m_chaining;
// true if this actor receives EXIT messages as ordinary messages
bool m_trap_exit;
// true if this actor takes part in cooperative scheduling
bool m_is_scheduled;
// pointer to the actor that is marked as successor due to a chained send
actor_ptr m_chained_actor;
// identifies the ID of the last sent synchronous request
message_id_t m_last_request_id;
message_id m_last_request_id;
// identifies all IDs of sync messages waiting for a response
std::vector<message_id_t> m_pending_responses;
std::vector<message_id> m_pending_responses;
// "default value" for m_current_node
detail::recursive_queue_node m_dummy_node;
// points to m_dummy_node if no callback is currently invoked,
// points to the node under processing otherwise
detail::recursive_queue_node* m_current_node;
// {group => subscription} map of all joined groups
std::map<group_ptr, group::subscription> m_subscriptions;
// allows actors to keep previous behaviors and enables unbecome()
detail::behavior_stack m_bhvr_stack;
inline void remove_handler(message_id_t id) {
m_bhvr_stack.erase(id);
}
void cleanup(std::uint32_t reason);
private:
std::function<void()> m_sync_failure_handler;
std::function<void()> m_sync_timeout_handler;
# endif // CPPA_DOCUMENTATION
protected:
virtual void do_become(behavior&& bhvr, bool discard_old) = 0;
inline void do_become(const behavior& bhvr, bool discard_old) {
behavior copy{bhvr};
do_become(std::move(copy), discard_old);
}
/** @endcond */
};
......@@ -457,10 +356,21 @@ class local_actor : public extend<actor,local_actor>::with<memory_cached> {
*/
typedef intrusive_ptr<local_actor> local_actor_ptr;
/******************************************************************************
* inline and template member function implementations *
******************************************************************************/
void local_actor::dequeue(behavior&& bhvr) {
behavior tmp{std::move(bhvr)};
dequeue(tmp);
}
inline void local_actor::dequeue_response(behavior&& bhvr, message_id id) {
behavior tmp{std::move(bhvr)};
dequeue_response(tmp, id);
}
inline bool local_actor::trap_exit() const {
return m_trap_exit;
}
......@@ -485,28 +395,90 @@ inline actor_ptr& local_actor::last_sender() {
return m_current_node->sender;
}
template<bool Discard>
inline void local_actor::become(behavior_policy<Discard>, behavior bhvr) {
do_become(std::move(bhvr), Discard);
template<bool Discard, typename... Ts>
inline void local_actor::become(behavior_policy<Discard>, Ts&&... args) {
do_become(match_expr_convert(std::forward<Ts>(args)...), Discard);
}
template<bool Discard, typename T0, typename T1, typename... Ts>
inline void local_actor::become(behavior_policy<Discard>, T0&& a0, T1&& a1, Ts&&... as) {
auto bhvr = match_expr_convert(std::forward<T0>(a0), std::forward<T1>(a1), std::forward<Ts>(as)...);
do_become(std::move(bhvr), Discard);
template<typename T, typename... Ts>
inline void local_actor::become(T arg, Ts&&... args) {
do_become(match_expr_convert(arg, std::forward<Ts>(args)...), true);
}
inline void local_actor::become(behavior bhvr) {
do_become(std::move(bhvr), true);
inline void local_actor::unbecome() {
m_bhvr_stack.pop_async_back();
}
template<typename... Cases, typename... Ts>
inline void local_actor::become(match_expr<Cases...> a, Ts&&... as) {
do_become(match_expr_convert(std::move(a), std::forward<Ts>(as)...), true);
inline bool local_actor::chaining_enabled() {
return m_chaining && !m_chained_actor;
}
inline void local_actor::unbecome() {
m_bhvr_stack.pop_async_back();
inline void local_actor::send_message(channel* whom, any_tuple&& what) {
whom->enqueue(this, std::move(what));
}
inline void local_actor::send_message(actor* whom, any_tuple&& what) {
if (chaining_enabled()) {
if (whom->chained_enqueue(this, std::move(what))) {
m_chained_actor.reset(whom);
}
}
else whom->enqueue(this, std::move(what));
}
inline message_id local_actor::send_sync_message(const actor_ptr& whom, any_tuple&& what) {
auto id = ++m_last_request_id;
CPPA_REQUIRE(id.is_request());
if (chaining_enabled()) {
if (whom->chained_sync_enqueue(this, id, std::move(what))) {
chained_actor(whom);
}
}
else whom->sync_enqueue(this, id, std::move(what));
auto awaited_response = id.response_id();
m_pending_responses.push_back(awaited_response);
return awaited_response;
}
inline message_id local_actor::get_response_id() {
auto id = m_current_node->mid;
return (id.is_request()) ? id.response_id() : message_id();
}
inline const actor_ptr& local_actor::chained_actor() {
return m_chained_actor;
}
inline void local_actor::chained_actor(const actor_ptr& value) {
m_chained_actor = value;
}
inline bool local_actor::awaits(message_id response_id) {
CPPA_REQUIRE(response_id.is_response());
return std::any_of(m_pending_responses.begin(),
m_pending_responses.end(),
[=](message_id other) {
return response_id == other;
});
}
inline void local_actor::mark_arrived(message_id response_id) {
auto last = m_pending_responses.end();
auto i = std::find(m_pending_responses.begin(), last, response_id);
if (i != last) m_pending_responses.erase(i);
}
inline detail::behavior_stack& local_actor::bhvr_stack() {
return m_bhvr_stack;
}
inline void local_actor::do_become(const behavior& bhvr, bool discard_old) {
behavior copy{bhvr};
do_become(std::move(copy), discard_old);
}
inline void local_actor::remove_handler(message_id id) {
m_bhvr_stack.erase(id);
}
} // namespace cppa
......
......@@ -56,7 +56,7 @@ class message_future {
public:
inline continue_helper(message_id_t mid) : m_mid(mid) { }
inline continue_helper(message_id mid) : m_mid(mid) { }
template<typename F>
void continue_with(F fun) {
......@@ -71,7 +71,7 @@ class message_future {
private:
message_id_t m_mid;
message_id m_mid;
};
......@@ -127,16 +127,16 @@ class message_future {
/**
* @brief Returns the awaited response ID.
*/
inline const message_id_t& id() const { return m_mid; }
inline const message_id& id() const { return m_mid; }
message_future(const message_future&) = default;
message_future& operator=(const message_future&) = default;
inline message_future(const message_id_t& from) : m_mid(from) { }
inline message_future(const message_id& from) : m_mid(from) { }
private:
message_id_t m_mid;
message_id m_mid;
template<typename... Fs>
behavior fs2bhvr(Fs... fs) {
......@@ -169,9 +169,9 @@ class sync_handle_helper {
inline sync_handle_helper(const message_future& mf) : m_mf(mf) { }
template<typename... Args>
inline message_future::continue_helper operator()(Args&&... args) {
return m_mf.then(std::forward<Args>(args)...);
template<typename... Ts>
inline message_future::continue_helper operator()(Ts&&... args) {
return m_mf.then(std::forward<Ts>(args)...);
}
private:
......
......@@ -42,7 +42,7 @@ struct invalid_message_id { constexpr invalid_message_id() { } };
* @brief
* @note Asynchronous messages always have an invalid message id.
*/
class message_id_t : util::comparable<message_id_t> {
class message_id : util::comparable<message_id> {
static constexpr std::uint64_t response_flag_mask = 0x8000000000000000;
static constexpr std::uint64_t answered_flag_mask = 0x4000000000000000;
......@@ -50,14 +50,14 @@ class message_id_t : util::comparable<message_id_t> {
public:
constexpr message_id_t() : m_value(0) { }
constexpr message_id() : m_value(0) { }
message_id_t(message_id_t&&) = default;
message_id_t(const message_id_t&) = default;
message_id_t& operator=(message_id_t&&) = default;
message_id_t& operator=(const message_id_t&) = default;
message_id(message_id&&) = default;
message_id(const message_id&) = default;
message_id& operator=(message_id&&) = default;
message_id& operator=(const message_id&) = default;
inline message_id_t& operator++() {
inline message_id& operator++() {
++m_value;
return *this;
}
......@@ -78,12 +78,12 @@ class message_id_t : util::comparable<message_id_t> {
return valid() && !is_response();
}
inline message_id_t response_id() const {
return message_id_t(valid() ? m_value | response_flag_mask : 0);
inline message_id response_id() const {
return message_id(valid() ? m_value | response_flag_mask : 0);
}
inline message_id_t request_id() const {
return message_id_t(m_value & request_id_mask);
inline message_id request_id() const {
return message_id(m_value & request_id_mask);
}
inline void mark_as_answered() {
......@@ -94,24 +94,24 @@ class message_id_t : util::comparable<message_id_t> {
return m_value;
}
static inline message_id_t from_integer_value(std::uint64_t value) {
message_id_t result;
static inline message_id from_integer_value(std::uint64_t value) {
message_id result;
result.m_value = value;
return result;
}
static constexpr invalid_message_id invalid = invalid_message_id{};
constexpr message_id_t(invalid_message_id) : m_value(0) { }
constexpr message_id(invalid_message_id) : m_value(0) { }
long compare(const message_id_t& other) const {
long compare(const message_id& other) const {
return (m_value == other.m_value)
? 0 : ((m_value < other.m_value) ? -1 : 1);
}
private:
explicit constexpr message_id_t(std::uint64_t value) : m_value(value) { }
explicit constexpr message_id(std::uint64_t value) : m_value(value) { }
std::uint64_t m_value;
......
......@@ -36,8 +36,6 @@
#include "cppa/config.hpp"
#include "cppa/ref_counted.hpp"
#include "cppa/network/protocol.hpp"
namespace cppa { namespace network {
class middleman;
......
......@@ -59,11 +59,11 @@ class sync_request_info : public extend<memory_managed,sync_request_info>::
pointer next; // intrusive next pointer
actor_ptr sender; // points to the sender of the message
message_id_t mid; // sync message ID
message_id mid; // sync message ID
private:
sync_request_info(actor_ptr sptr, message_id_t id);
sync_request_info(actor_ptr sptr, message_id id);
};
......@@ -79,7 +79,7 @@ class default_actor_proxy : public actor_proxy {
void enqueue(const actor_ptr& sender, any_tuple msg);
void sync_enqueue(const actor_ptr& sender, message_id_t id, any_tuple msg);
void sync_enqueue(const actor_ptr& sender, message_id id, any_tuple msg);
void link_to(const actor_ptr& other);
......@@ -107,7 +107,7 @@ class default_actor_proxy : public actor_proxy {
void forward_msg(const actor_ptr& sender,
any_tuple msg,
message_id_t mid = message_id_t());
message_id mid = message_id());
default_protocol_ptr m_proto;
process_information_ptr m_pinf;
......
......@@ -49,13 +49,13 @@ class message_header {
actor_ptr sender;
actor_ptr receiver;
message_id_t id;
message_id id;
message_header();
message_header(const actor_ptr& sender,
const actor_ptr& receiver,
message_id_t id = message_id_t::invalid);
message_id id = message_id::invalid);
inline void deliver(any_tuple msg) const {
if (receiver) {
......
......@@ -36,15 +36,16 @@
#include <memory>
#include <functional>
#include "cppa/network/protocol.hpp"
#include "cppa/network/acceptor.hpp"
#include "cppa/network/continuable_reader.hpp"
#include "cppa/network/continuable_io.hpp"
namespace cppa { namespace detail { class singleton_manager; } }
namespace cppa { namespace network {
class protocol;
typedef intrusive_ptr<protocol> protocol_ptr;
/**
* @brief Multiplexes asynchronous IO.
*/
......
......@@ -39,9 +39,9 @@
#include "cppa/actor.hpp"
#include "cppa/ref_counted.hpp"
#include "cppa/primitive_variant.hpp"
#include "cppa/intrusive_fwd_ptr.hpp"
#include "cppa/network/acceptor.hpp"
#include "cppa/network/middleman.hpp"
namespace cppa { class actor_addressing; }
......@@ -83,14 +83,6 @@ class protocol : public ref_counted {
void run_later(std::function<void()> fun);
struct ref_ftor {
void operator()(abstract_middleman*) const;
};
struct deref_ftor {
void operator()(abstract_middleman*) const;
};
protected:
// note: not thread-safe; call only in run_later functor!
......@@ -111,7 +103,7 @@ class protocol : public ref_counted {
private:
intrusive_fwd_ptr<abstract_middleman,ref_ftor,deref_ftor> m_parent;
intrusive_ptr<abstract_middleman> m_parent;
};
......
......@@ -119,7 +119,7 @@ class object {
/**
* @brief Creates a (deep) copy of @p other and assigns it to @p this.
* @return @p *this
* @returns @p *this
*/
object& operator=(const object& other);
......
......@@ -61,60 +61,54 @@ class partial_function {
public:
/** @cond PRIVATE */
typedef intrusive_ptr<detail::behavior_impl> impl_ptr;
inline auto as_behavior_impl() const -> const impl_ptr&;
partial_function(impl_ptr ptr);
static constexpr bool may_have_timeout = false;
/** @endcond */
partial_function() = default;
partial_function(partial_function&&) = default;
partial_function(const partial_function&) = default;
partial_function& operator=(partial_function&&) = default;
partial_function& operator=(const partial_function&) = default;
template<typename... Cases>
partial_function(const match_expr<Cases...>& mexpr)
: m_impl(mexpr.as_behavior_impl()) { }
partial_function(impl_ptr ptr);
inline bool undefined() const {
return m_impl == nullptr;
}
inline bool defined_at(const any_tuple& value) {
return (m_impl) && m_impl->defined_at(value);
}
template<typename T>
inline bool operator()(T&& arg) {
return (m_impl) && m_impl->invoke(std::forward<T>(arg));
}
inline partial_function or_else(const partial_function& other) const {
return m_impl->or_else(other.m_impl);
}
inline behavior or_else(const behavior& other) const {
return behavior{m_impl->or_else(other.m_impl)};
}
template<typename... Cs>
partial_function(const match_expr<Cs...>& mexpr);
/**
* @brief Returns @p true if this partial function is defined for the
* types of @p value, false otherwise.
*/
inline bool defined_at(const any_tuple& value);
/**
* @brief Returns @p true if this partial function was applied to
* @p args, false otherwise.
* @note This member function can return @p false even if
* {@link defined_at()} returns @p true, because {@link defined_at()}
* does <b>not</b> evaluate guards.
*/
template<typename T>
typename std::conditional<T::may_have_timeout,behavior,partial_function>::type
or_else(const T& arg) const;
inline bool operator()(T&& arg);
template<typename T0, typename T1, typename... Ts>
/**
* @brief Adds a fallback which is used where
* this partial function is not defined.
*/
template<typename... Ts>
typename std::conditional<
util::disjunction<
T0::may_have_timeout,
T1::may_have_timeout,
Ts::may_have_timeout...
>::value,
util::disjunction<Ts::may_have_timeout...>::value,
behavior,
partial_function
>::type
or_else(const T0& arg0, const T1& arg1, const Ts&... args) const;
inline const impl_ptr& as_behavior_impl() const;
or_else(Ts&&... args) const;
private:
......@@ -122,45 +116,57 @@ class partial_function {
};
template<typename T, typename... Ts>
template<typename T>
typename std::conditional<T::may_have_timeout,behavior,partial_function>::type
match_expr_convert(const T& arg) {
return {arg};
}
template<typename T0, typename T1, typename... Ts>
typename std::conditional<
util::disjunction<
T::may_have_timeout,
T0::may_have_timeout,
T1::may_have_timeout,
Ts::may_have_timeout...
>::value,
behavior,
partial_function
>::type
match_expr_convert(const T& arg0, const Ts&... args) {
return detail::match_expr_concat(arg0, args...);
match_expr_convert(const T0& arg0, const T1& arg1, const Ts&... args) {
return detail::match_expr_concat(arg0, arg1, args...);
}
/******************************************************************************
* inline and template member function implementations *
******************************************************************************/
inline const partial_function::impl_ptr& partial_function::as_behavior_impl() const {
return m_impl;
template<typename... Cs>
partial_function::partial_function(const match_expr<Cs...>& mexpr)
: m_impl(mexpr.as_behavior_impl()) { }
inline bool partial_function::defined_at(const any_tuple& value) {
return (m_impl) && m_impl->defined_at(value);
}
template<typename T>
typename std::conditional<T::may_have_timeout,behavior,partial_function>::type
partial_function::or_else(const T& arg) const {
return m_impl->or_else(arg.as_behavior_impl());
inline bool partial_function::operator()(T&& arg) {
return (m_impl) && m_impl->invoke(std::forward<T>(arg));
}
template<typename T0, typename T1, typename... Ts>
template<typename... Ts>
typename std::conditional<
util::disjunction<
T0::may_have_timeout,
T1::may_have_timeout,
Ts::may_have_timeout...
>::value,
util::disjunction<Ts::may_have_timeout...>::value,
behavior,
partial_function
>::type
partial_function::or_else(const T0& arg0, const T1& arg1, const Ts&... args) const {
return m_impl->or_else(match_expr_convert(arg0, arg1, args...).as_behavior_impl());
partial_function::or_else(Ts&&... args) const {
auto tmp = match_expr_convert(std::forward<Ts>(args)...);
return m_impl->or_else(tmp.as_behavior_impl());
}
inline auto partial_function::as_behavior_impl() const -> const impl_ptr& {
return m_impl;
}
} // namespace cppa
......
......@@ -38,62 +38,64 @@
namespace cppa {
#ifdef CPPA_DOCUMENTATION
/**
* @ingroup BlockingAPI
* @{
*/
/**
* @brief Dequeues the next message from the mailbox that's matched
* by @p bhvr and executes the corresponding callback.
* @param bhvr Denotes the actor's response the next incoming message.
* @brief Dequeues the next message from the mailbox that
* is matched by given behavior.
*/
void receive(behavior& bhvr);
template<typename... Ts>
void receive(Ts&&... args);
/**
* @brief Receives messages in an endless loop.
*
* Semantically equal to: <tt>for (;;) { receive(bhvr); }</tt>
* @param bhvr Denotes the actor's response the next incoming message.
* Semantically equal to: <tt>for (;;) { receive(...); }</tt>
*/
void receive_loop(behavior& bhvr);
template<typename... Ts>
void receive_loop(Ts&&... args);
/**
* @brief Receives messages as long as @p stmt returns true.
* @brief Receives messages as in a range-based loop.
*
* Semantically equal to: <tt>while (stmt()) { receive(...); }</tt>.
* Semantically equal to:
* <tt>for ( ; begin != end; ++begin) { receive(...); }</tt>.
*
* <b>Usage example:</b>
* @code
* int i = 0;
* receive_while([&]() { return (++i <= 10); })
* (
* on<int>() >> int_fun,
* on<float>() >> float_fun
* receive_for(i, 10) (
* on(atom("get")) >> [&]() { reply("result", i); }
* );
* @endcode
* @param stmt Lambda expression, functor or function returning a @c bool.
* @param begin First value in range.
* @param end Last value in range (excluded).
* @returns A functor implementing the loop.
*/
template<typename Statement>
auto receive_while(Statement&& stmt);
template<typename T>
detail::receive_for_helper<T> receive_for(T& begin, const T& end);
/**
* @brief Receives messages as in a range-based loop.
* @brief Receives messages as long as @p stmt returns true.
*
* Semantically equal to: <tt>for ( ; begin != end; ++begin) { receive(...); }</tt>.
* Semantically equal to: <tt>while (stmt()) { receive(...); }</tt>.
*
* <b>Usage example:</b>
* @code
* int i = 0;
* receive_for(i, 10)
* receive_while([&]() { return (++i <= 10); })
* (
* on(atom("get")) >> [&]() { reply("result", i); }
* on<int>() >> int_fun,
* on<float>() >> float_fun
* );
* @endcode
* @param begin First value in range.
* @param end Last value in range (excluded).
* @param stmt Lambda expression, functor or function returning a @c bool.
* @returns A functor implementing the loop.
*/
template<typename T>
auto receive_for(T& begin, const T& end);
template<typename Statement>
detail::receive_while_helper<Statement> receive_while(Statement&& stmt);
/**
* @brief Receives messages until @p stmt returns true.
......@@ -113,32 +115,34 @@ auto receive_for(T& begin, const T& end);
* @param bhvr Denotes the actor's response the next incoming message.
* @returns A functor providing the @c until member function.
*/
auto do_receive(behavior& bhvr);
template<typename... Args>
detail::do_receive_helper do_receive(Args&&... args);
#else // CPPA_DOCUMENTATION
/**
* @}
*/
inline void receive(behavior& bhvr) { self->dequeue(bhvr); }
inline void receive(partial_function& fun) { self->dequeue(fun); }
/******************************************************************************
* inline and template function implementations *
******************************************************************************/
template<typename Arg0, typename... Args>
void receive(Arg0&& arg0, Args&&... args) {
auto tmp = match_expr_convert(std::forward<Arg0>(arg0),
std::forward<Args>(args)...);
receive(tmp);
template<typename... Ts>
void receive(Ts&&... args) {
static_assert(sizeof...(Ts), "receive() requires at least one argument");
self->dequeue(match_expr_convert(std::forward<Ts>(args)...));
}
void receive_loop(behavior& rules);
void receive_loop(behavior&& rules);
void receive_loop(partial_function& rules);
void receive_loop(partial_function&& rules);
inline void receive_loop(behavior rules) {
local_actor* sptr = self;
for (;;) sptr->dequeue(rules);
}
template<typename Arg0, typename... Args>
void receive_loop(Arg0&& arg0, Args&&... args) {
auto tmp = match_expr_convert(std::forward<Arg0>(arg0),
std::forward<Args>(args)...);
receive_loop(tmp);
template<typename... Ts>
void receive_loop(Ts&&... args) {
behavior bhvr = match_expr_convert(std::forward<Ts>(args)...);
local_actor* sptr = self;
for (;;) sptr->dequeue(bhvr);
}
template<typename T>
......@@ -153,13 +157,12 @@ detail::receive_while_helper<Statement> receive_while(Statement&& stmt) {
return std::move(stmt);
}
template<typename... Args>
detail::do_receive_helper do_receive(Args&&... args) {
return detail::do_receive_helper(std::forward<Args>(args)...);
template<typename... Ts>
detail::do_receive_helper do_receive(Ts&&... args) {
behavior bhvr = match_expr_convert(std::forward<Ts>(args)...);
return detail::do_receive_helper(std::move(bhvr));
}
#endif // CPPA_DOCUMENTATION
} // namespace cppa
#endif // CPPA_RECEIVE_HPP
......@@ -52,7 +52,7 @@ class response_handle {
response_handle(const actor_ptr& from,
const actor_ptr& to,
const message_id_t& response_id);
const message_id& response_id);
/**
* @brief Queries whether response message is still outstanding.
......@@ -73,7 +73,7 @@ class response_handle {
/**
* @brief Returns the message id for the response message.
*/
inline const message_id_t& response_id() const { return m_id; }
inline const message_id& response_id() const { return m_id; }
/**
* @brief Returns the actor that is going send the response message.
......@@ -89,7 +89,7 @@ class response_handle {
actor_ptr m_from;
actor_ptr m_to;
message_id_t m_id;
message_id m_id;
};
......
......@@ -119,9 +119,9 @@ class scheduled_actor : public extend<local_actor,scheduled_actor>::with<mailbox
bool chained_enqueue(const actor_ptr&, any_tuple);
void sync_enqueue(const actor_ptr&, message_id_t, any_tuple);
void sync_enqueue(const actor_ptr&, message_id, any_tuple);
bool chained_sync_enqueue(const actor_ptr&, message_id_t, any_tuple);
bool chained_sync_enqueue(const actor_ptr&, message_id, any_tuple);
void request_timeout(const util::duration& d);
......@@ -195,7 +195,7 @@ class scheduled_actor : public extend<local_actor,scheduled_actor>::with<mailbox
bool sync_enqueue_impl(actor_state next,
const actor_ptr& sender,
any_tuple& msg,
message_id_t id);
message_id id);
bool m_has_pending_tout;
std::uint32_t m_pending_tout;
......
......@@ -144,7 +144,7 @@ class scheduler {
template<typename Duration, typename... Data>
void delayed_reply(const actor_ptr& to,
const Duration& rel_time,
message_id_t id,
message_id id,
any_tuple data ) {
CPPA_REQUIRE(!id.valid() || id.is_response());
if (id.valid()) {
......
......@@ -61,15 +61,11 @@ class stacked : public Base {
static constexpr auto receive_flag = detail::rp_nestable;
virtual void dequeue(partial_function& fun) {
m_recv_policy.receive(dthis(), fun);
}
virtual void dequeue(behavior& bhvr) {
m_recv_policy.receive(dthis(), bhvr);
}
virtual void dequeue_response(behavior& bhvr, message_id_t request_id) {
virtual void dequeue_response(behavior& bhvr, message_id request_id) {
m_recv_policy.receive(dthis(), bhvr, request_id);
}
......@@ -98,10 +94,10 @@ class stacked : public Base {
: Base(std::forward<Args>(args)...), m_behavior(std::move(fun)) { }
virtual void do_become(behavior&& bhvr, bool discard_old) {
become_impl(std::move(bhvr), discard_old, message_id_t());
become_impl(std::move(bhvr), discard_old, message_id());
}
virtual void become_waiting_for(behavior&& bhvr, message_id_t mid) {
virtual void become_waiting_for(behavior bhvr, message_id mid) {
become_impl(std::move(bhvr), false, mid);
}
......@@ -123,7 +119,7 @@ class stacked : public Base {
std::function<void()> m_behavior;
detail::receive_policy m_recv_policy;
void become_impl(behavior&& bhvr, bool discard_old, message_id_t mid) {
void become_impl(behavior&& bhvr, bool discard_old, message_id mid) {
if (bhvr.timeout().valid()) {
dthis()->reset_timeout();
dthis()->request_timeout(bhvr.timeout());
......
......@@ -90,7 +90,7 @@ class thread_mapped_actor : public extend<local_actor,thread_mapped_actor>::with
void enqueue(const actor_ptr& sender, any_tuple msg);
void sync_enqueue(const actor_ptr& sender, message_id_t id, any_tuple msg);
void sync_enqueue(const actor_ptr& sender, message_id id, any_tuple msg);
inline void reset_timeout() { }
inline void request_timeout(const util::duration&) { }
......
......@@ -69,7 +69,7 @@ bool actor::chained_enqueue(const actor_ptr& sender, any_tuple msg) {
return false;
}
bool actor::chained_sync_enqueue(const actor_ptr& ptr, message_id_t id, any_tuple msg) {
bool actor::chained_sync_enqueue(const actor_ptr& ptr, message_id id, any_tuple msg) {
sync_enqueue(ptr, id, std::move(msg));
return false;
}
......
......@@ -62,7 +62,7 @@ struct behavior_stack_mover : iterator<output_iterator_tag,void,void,void,void>{
inline behavior_stack_mover move_iter(behavior_stack* bs) { return {bs}; }
option<behavior&> behavior_stack::sync_handler(message_id_t expected_response) {
option<behavior&> behavior_stack::sync_handler(message_id expected_response) {
if (expected_response.valid()) {
auto e = m_elements.rend();
auto i = find_if(m_elements.rbegin(), e, [=](element_type& val) {
......
......@@ -42,11 +42,11 @@ using namespace std;
namespace cppa { namespace network {
inline sync_request_info* new_req_info(actor_ptr sptr, message_id_t id) {
inline sync_request_info* new_req_info(actor_ptr sptr, message_id id) {
return detail::memory::create<sync_request_info>(std::move(sptr), id);
}
sync_request_info::sync_request_info(actor_ptr sptr, message_id_t id)
sync_request_info::sync_request_info(actor_ptr sptr, message_id id)
: next(nullptr), sender(std::move(sptr)), mid(id) { }
default_actor_proxy::default_actor_proxy(actor_id mid,
......@@ -89,7 +89,7 @@ void default_actor_proxy::deliver(const network::message_header& hdr, any_tuple
void default_actor_proxy::forward_msg(const actor_ptr& sender,
any_tuple msg,
message_id_t mid) {
message_id mid) {
CPPA_LOG_TRACE("");
if (sender && mid.is_request()) {
switch (m_pending_requests.enqueue(new_req_info(sender, mid))) {
......@@ -147,7 +147,7 @@ void default_actor_proxy::enqueue(const actor_ptr& sender, any_tuple msg) {
}
void default_actor_proxy::sync_enqueue(const actor_ptr& sender,
message_id_t mid,
message_id mid,
any_tuple msg) {
CPPA_LOG_TRACE(CPPA_TARG(sender, to_string) << ", "
<< CPPA_MARG(mid, integer_value) << ", "
......
......@@ -78,11 +78,7 @@ void event_based_actor::dequeue(behavior&) {
quit(exit_reason::unallowed_function_call);
}
void event_based_actor::dequeue(partial_function&) {
quit(exit_reason::unallowed_function_call);
}
void event_based_actor::dequeue_response(behavior&, message_id_t) {
void event_based_actor::dequeue_response(behavior&, message_id) {
quit(exit_reason::unallowed_function_call);
}
......@@ -193,7 +189,7 @@ void event_based_actor::do_become(behavior&& bhvr, bool discard_old) {
m_bhvr_stack.push_back(std::move(bhvr));
}
void event_based_actor::become_waiting_for(behavior&& bhvr, message_id_t mf) {
void event_based_actor::become_waiting_for(behavior bhvr, message_id mf) {
if (bhvr.timeout().valid()) {
reset_timeout();
request_timeout(bhvr.timeout());
......
......@@ -75,11 +75,11 @@ 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) { }
void local_actor::monitor(actor_ptr whom) {
void local_actor::monitor(const actor_ptr& whom) {
if (whom) whom->attach(attachable_ptr{new down_observer(this, whom)});
}
void local_actor::demonitor(actor_ptr whom) {
void local_actor::demonitor(const actor_ptr& whom) {
attachable::token mtoken{typeid(down_observer), this};
if (whom) whom->detach(mtoken);
}
......@@ -113,12 +113,12 @@ void local_actor::reply_message(any_tuple&& what) {
}
auto& id = m_current_node->mid;
if (id.valid() == false || id.is_response()) {
send_message(whom, std::move(what));
send_message(whom.get(), std::move(what));
}
else if (!id.is_answered()) {
if (chaining_enabled()) {
if (whom->chained_sync_enqueue(this, id.response_id(), std::move(what))) {
m_chained_actor = whom;
chained_actor(whom);
}
}
else {
......@@ -140,7 +140,7 @@ void local_actor::forward_message(const actor_ptr& new_receiver) {
else {
new_receiver->sync_enqueue(from.get(), id, m_current_node->msg);
// treat this message as asynchronous message from now on
m_current_node->mid = message_id_t();
m_current_node->mid = message_id();
}
}
......@@ -151,7 +151,7 @@ response_handle local_actor::make_response_handle() {
return std::move(result);
}
message_id_t local_actor::send_timed_sync_message(const actor_ptr& whom,
message_id local_actor::send_timed_sync_message(const actor_ptr& whom,
const util::duration& rel_time,
any_tuple&& what) {
auto mid = this->send_sync_message(whom, std::move(what));
......@@ -164,10 +164,6 @@ void local_actor::exec_behavior_stack() {
// default implementation does nothing
}
sync_handle_helper local_actor::handle_response(const message_future& mf) {
return ::cppa::handle_response(mf);
}
void local_actor::cleanup(std::uint32_t reason) {
m_subscriptions.clear();
super::cleanup(reason);
......
......@@ -34,7 +34,7 @@ namespace cppa { namespace network {
message_header::message_header(const actor_ptr& s,
const actor_ptr& r,
message_id_t mid )
message_id mid )
: sender(s), receiver(r), id(mid) { }
message_header::message_header() : sender(nullptr), receiver(nullptr), id() { }
......
......@@ -34,14 +34,6 @@
namespace cppa { namespace network {
void protocol::ref_ftor::operator()(abstract_middleman* ptr) const {
if (ptr) ptr->ref();
}
void protocol::deref_ftor::operator()(abstract_middleman* ptr) const {
if (ptr) ptr->deref();
}
protocol::protocol(abstract_middleman* parent) : m_parent(parent) {
CPPA_REQUIRE(parent != nullptr);
}
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011-2013 *
* Dominik Charousset <dominik.charousset@haw-hamburg.de> *
* *
* This file is part of libcppa. *
* libcppa is free software: you can redistribute it and/or modify it under *
* the terms of the GNU Lesser General Public License as published by the *
* Free Software Foundation, either version 3 of the License *
* or (at your option) any later version. *
* *
* libcppa is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. *
* See the GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with libcppa. If not, see <http://www.gnu.org/licenses/>. *
\******************************************************************************/
#include "cppa/cppa.hpp"
#include "cppa/receive.hpp"
namespace cppa {
void receive_loop(behavior& rules) {
local_actor* sptr = self;
for (;;) {
sptr->dequeue(rules);
}
}
void receive_loop(behavior&& rules) {
behavior tmp(std::move(rules));
receive_loop(tmp);
}
void receive_loop(partial_function& rules) {
local_actor* sptr = self;
for (;;) {
sptr->dequeue(rules);
}
}
void receive_loop(partial_function&& rules) {
partial_function tmp(std::move(rules));
receive_loop(tmp);
}
} // namespace cppa
......@@ -34,7 +34,7 @@ namespace cppa { namespace detail {
recursive_queue_node::recursive_queue_node(const actor_ptr& sptr,
any_tuple data,
message_id_t id)
message_id id)
: next(nullptr), marked(false), sender(sptr), msg(std::move(data)), mid(id) { }
} } // namespace cppa::detail
......@@ -40,7 +40,7 @@ namespace cppa {
response_handle::response_handle(const actor_ptr& from,
const actor_ptr& to,
const message_id_t& id)
const message_id& id)
: m_from(from), m_to(to), m_id(id) {
CPPA_REQUIRE(id.is_response());
}
......
......@@ -133,7 +133,7 @@ bool scheduled_actor::enqueue(actor_state next_state,
bool scheduled_actor::sync_enqueue_impl(actor_state next,
const actor_ptr& sender,
any_tuple& msg,
message_id_t id) {
message_id id) {
bool err = false;
auto ptr = new_mailbox_element(sender, std::move(msg), id);
bool res = enqueue(next, &err, ptr);
......@@ -162,13 +162,13 @@ bool scheduled_actor::chained_enqueue(const actor_ptr& sender, any_tuple msg) {
}
void scheduled_actor::sync_enqueue(const actor_ptr& sender,
message_id_t id,
message_id id,
any_tuple msg) {
sync_enqueue_impl(actor_state::ready, sender, msg, id);
}
bool scheduled_actor::chained_sync_enqueue(const actor_ptr& sender,
message_id_t id,
message_id id,
any_tuple msg) {
return sync_enqueue_impl(actor_state::pending, sender, msg, id);
}
......
......@@ -38,14 +38,13 @@ scheduled_actor_dummy::scheduled_actor_dummy()
void scheduled_actor_dummy::enqueue(const actor_ptr&, any_tuple) { }
void scheduled_actor_dummy::quit(std::uint32_t) { }
void scheduled_actor_dummy::dequeue(behavior&) { }
void scheduled_actor_dummy::dequeue(partial_function&) { }
void scheduled_actor_dummy::dequeue_response(behavior&, message_id_t) { }
void scheduled_actor_dummy::dequeue_response(behavior&, message_id) { }
void scheduled_actor_dummy::do_become(behavior&&, bool) { }
void scheduled_actor_dummy::become_waiting_for(behavior&&, message_id_t) { }
void scheduled_actor_dummy::become_waiting_for(behavior, message_id) { }
bool scheduled_actor_dummy::has_behavior() { return false; }
void scheduled_actor_dummy::sync_enqueue(const actor_ptr&,
message_id_t,
message_id,
any_tuple) { }
resume_result scheduled_actor_dummy::resume(util::fiber*,actor_ptr&) {
......
......@@ -68,13 +68,13 @@ class delayed_msg {
delayed_msg(const channel_ptr& arg0,
const actor_ptr& arg1,
message_id_t,
message_id,
any_tuple&& arg3)
: ptr_a(arg0), from(arg1), msg(move(arg3)) { }
delayed_msg(const actor_ptr& arg0,
const actor_ptr& arg1,
message_id_t arg2,
message_id arg2,
any_tuple&& arg3)
: ptr_b(arg0), from(arg1), id(arg2), msg(move(arg3)) { }
......@@ -94,7 +94,7 @@ class delayed_msg {
channel_ptr ptr_a;
actor_ptr ptr_b;
actor_ptr from;
message_id_t id;
message_id id;
any_tuple msg;
};
......@@ -146,7 +146,7 @@ inline void insert_dmsg(Map& storage,
const T& to,
const actor_ptr& sender,
any_tuple&& tup,
message_id_t id = message_id_t{}) {
message_id id = message_id{}) {
auto tout = hrc::now();
tout += d;
delayed_msg dmsg{to, sender, id, move(tup)};
......@@ -170,7 +170,7 @@ void scheduler_helper::timer_loop(scheduler_helper::ptr_type m_self) {
},
on(atom("REPLY"), arg_match) >> [&](const util::duration& d,
const actor_ptr& ptr,
message_id_t id,
message_id id,
any_tuple& tup) {
insert_dmsg(messages, d, ptr, msg_ptr->sender, move(tup), id);
},
......
......@@ -60,7 +60,7 @@ void thread_mapped_actor::enqueue(const actor_ptr& sender, any_tuple msg) {
}
void thread_mapped_actor::sync_enqueue(const actor_ptr& sender,
message_id_t id,
message_id id,
any_tuple msg) {
auto ptr = this->new_mailbox_element(sender, std::move(msg), id);
if (!this->m_mailbox.push_back(ptr)) {
......
......@@ -43,29 +43,27 @@
#include "cppa/atom.hpp"
#include "cppa/to_string.hpp"
#include "cppa/exception.hpp"
#include "cppa/singletons.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/binary_serializer.hpp"
#include "cppa/binary_deserializer.hpp"
#include "cppa/intrusive/single_reader_queue.hpp"
#include "cppa/network/acceptor.hpp"
#include "cppa/network/protocol.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/network/ipv4_acceptor.hpp"
#include "cppa/network/ipv4_io_stream.hpp"
#include "cppa/detail/actor_registry.hpp"
#include "cppa/detail/singleton_manager.hpp"
namespace cppa {
using namespace detail;
using namespace network;
namespace {
protocol* proto() {
return singleton_manager::get_middleman()->protocol(atom("DEFAULT")).get();
}
} // namespace <anonymous>
namespace { protocol_ptr proto() {
return get_middleman()->protocol(atom("DEFAULT")).get();
} }
void publish(actor_ptr whom, std::unique_ptr<acceptor> aptr) {
proto()->publish(whom, move(aptr), {});
......
......@@ -412,7 +412,7 @@ class msg_hdr_tinfo : public util::abstract_uniform_type_info<network::message_h
actor_ptr_tinfo::s_deserialize(msg.sender, source, actor_ptr_name);
actor_ptr_tinfo::s_deserialize(msg.receiver, source, actor_ptr_name);
auto msg_id = source->read<std::uint64_t>();
msg.id = message_id_t::from_integer_value(msg_id);
msg.id = message_id::from_integer_value(msg_id);
source->end_object();
}
......
......@@ -461,7 +461,7 @@ int main() {
self->become (
on("hi") >> [&]() {
auto handle = sync_send(self->last_sender(), "whassup?");
self->handle_response(handle) (
handle_response(handle) (
on_arg_match >> [&](const string& str) {
CPPA_CHECK(self->last_sender() != nullptr);
CPPA_CHECK_EQUAL(str, "nothing");
......
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