Commit e39416ae authored by neverlord's avatar neverlord

removed class message

parent 79c5acb1
......@@ -36,7 +36,6 @@ libcppa_la_SOURCES = \
src/invoke_rules.cpp \
src/mailman.cpp \
src/matcher_arguments.cpp \
src/message.cpp \
src/message_queue.cpp \
src/mock_scheduler.cpp \
src/native_socket.cpp \
......@@ -147,7 +146,6 @@ nobase_library_include_HEADERS = \
cppa/invoke.hpp \
cppa/invoke_rules.hpp \
cppa/match.hpp \
cppa/message.hpp \
cppa/message_queue.hpp \
cppa/object.hpp \
cppa/on.hpp \
......
......@@ -27,7 +27,6 @@ cppa/intrusive_ptr.hpp
unit_testing/test__spawn.cpp
src/mock_scheduler.cpp
cppa/actor.hpp
cppa/message.hpp
unit_testing/test__intrusive_ptr.cpp
cppa/util/concat_type_lists.hpp
cppa/util/filter_type_list.hpp
......@@ -132,7 +131,6 @@ src/binary_deserializer.cpp
src/actor.cpp
cppa/announce.hpp
src/abstract_type_list.cpp
src/message.cpp
cppa/util/shared_spinlock.hpp
src/shared_spinlock.cpp
cppa/util/shared_lock_guard.hpp
......
......@@ -15,13 +15,13 @@ class actor_proxy : public detail::abstract_actor<actor>
typedef detail::abstract_actor<actor> super;
// implemented in unicast_network.cpp
static void forward_message(const process_information_ptr&, const message&);
static void forward_message(const process_information_ptr&, const any_tuple&);
public:
actor_proxy(std::uint32_t mid, const process_information_ptr& parent);
void enqueue(const message& msg);
void enqueue(const any_tuple& msg);
void link_to(intrusive_ptr<actor>& other);
......
......@@ -63,6 +63,12 @@ class any_tuple
return size() == 0;
}
template<typename T>
inline const T& get_as(size_t p) const;
template<typename T>
inline T& get_mutable_as(size_t p);
};
inline bool operator==(const any_tuple& lhs, const any_tuple& rhs)
......@@ -75,6 +81,19 @@ inline bool operator!=(const any_tuple& lhs, const any_tuple& rhs)
return !(lhs == rhs);
}
template<typename T>
inline const T& any_tuple::get_as(size_t p) const
{
return *reinterpret_cast<const T*>(at(p));
}
template<typename T>
inline T& any_tuple::get_mutable_as(size_t p)
{
return *reinterpret_cast<T*>(mutable_at(p));
}
} // namespace cppa
#endif // ANY_TUPLE_HPP
......@@ -9,7 +9,7 @@ namespace cppa {
// forward declarations
class actor;
class group;
class message;
class any_tuple;
/**
* @brief Interface for all message receivers.
......@@ -34,7 +34,7 @@ class channel : public ref_counted
/**
* @brief Enqueues @p msg to the list of received messages.
*/
virtual void enqueue(const message& msg) = 0;
virtual void enqueue(const any_tuple& msg) = 0;
};
......
......@@ -40,7 +40,7 @@
#include "cppa/invoke.hpp"
#include "cppa/channel.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/message.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/announce.hpp"
#include "cppa/scheduler.hpp"
#include "cppa/to_string.hpp"
......@@ -346,7 +346,7 @@ detail::do_receive_helper do_receive(Args&&... args)
* @brief Gets the last dequeued message from the mailbox.
* @returns The last dequeued message from the mailbox.
*/
inline const message& last_received()
inline const any_tuple& last_received()
{
return self()->mailbox().last_dequeued();
}
......@@ -384,7 +384,7 @@ template<class C, typename Arg0, typename... Args>
typename util::enable_if<std::is_base_of<channel, C>, void>::type
send(intrusive_ptr<C>& whom, const Arg0& arg0, const Args&... args)
{
if (whom) whom->enqueue(message(self(), whom, arg0, args...));
if (whom) whom->enqueue(make_tuple(arg0, args...));
}
template<class C, typename Arg0, typename... Args>
......@@ -392,21 +392,21 @@ typename util::enable_if<std::is_base_of<channel, C>, void>::type
send(intrusive_ptr<C>&& whom, const Arg0& arg0, const Args&... args)
{
intrusive_ptr<C> tmp(std::move(whom));
if (tmp) tmp->enqueue(message(self(), whom, arg0, args...));
if (tmp) tmp->enqueue(make_tuple(arg0, args...));
}
// matches send(self(), ...);
template<typename Arg0, typename... Args>
void send(local_actor* whom, const Arg0& arg0, const Args&... args)
{
if (whom) whom->enqueue(message(self(), whom, arg0, args...));
if (whom) whom->enqueue(make_tuple(arg0, args...));
}
template<class C>
typename util::enable_if<std::is_base_of<channel, C>, intrusive_ptr<C>&>::type
operator<<(intrusive_ptr<C>& whom, const any_tuple& what)
{
if (whom) whom->enqueue(message(self(), whom, what));
if (whom) whom->enqueue(what);
return whom;
}
......@@ -415,7 +415,7 @@ typename util::enable_if<std::is_base_of<channel, C>, intrusive_ptr<C>>::type
operator<<(intrusive_ptr<C>&& whom, const any_tuple& what)
{
intrusive_ptr<C> tmp(std::move(whom));
if (tmp) tmp->enqueue(message(self(), tmp, what));
if (tmp) tmp->enqueue(what);
return std::move(tmp);
}
......@@ -423,7 +423,7 @@ template<class C>
typename util::enable_if<std::is_base_of<channel, C>, intrusive_ptr<C>&>::type
operator<<(intrusive_ptr<C>& whom, any_tuple&& what)
{
if (whom) whom->enqueue(message(self(), whom, std::move(what)));
if (whom) whom->enqueue(std::move(what));
return whom;
}
......@@ -432,7 +432,7 @@ typename util::enable_if<std::is_base_of<channel, C>, intrusive_ptr<C>>::type
operator<<(intrusive_ptr<C>&& whom, any_tuple&& what)
{
intrusive_ptr<C> tmp(std::move(whom));
if (tmp) tmp->enqueue(message(self(), tmp, std::move(what)));
if (tmp) tmp->enqueue(std::move(what));
return std::move(tmp);
}
......@@ -444,19 +444,6 @@ local_actor* operator<<(local_actor* whom, any_tuple&& what);
#endif
/**
* @brief Replies to the last received message.
* @param arg0 First value for the message content.
* @param args Any number of values for the message content.
*/
template<typename Arg0, typename... Args>
void reply(const Arg0& arg0, const Args&... args)
{
local_actor* sptr = self();
actor_ptr whom = sptr->mailbox().last_dequeued().sender();
if (whom) whom->enqueue(message(sptr, whom, arg0, args...));
}
/**
* @brief Sends a message to @p whom that is delayed by @p rel_time.
* @param whom Receiver of the message.
......@@ -469,21 +456,6 @@ void future_send(actor_ptr whom, const Duration& rel_time, const Data&... data)
get_scheduler()->future_send(self(), whom, rel_time, data...);
}
/**
* @brief Replies to the last received message with @p rel_time delay.
* @param rel_time Relative time duration to delay the message.
* @param data Any number of values for the message content.
*/
template<typename Duration, typename... Data>
void delayed_reply(const Duration& rel_time, const Data&... data)
{
auto whom = last_received().sender();
if (whom)
{
get_scheduler()->future_send(self(), whom, rel_time, data...);
}
}
/**
* @brief Blocks execution of this actor until all
* other actors finished execution.
......
......@@ -111,7 +111,7 @@ class abstract_actor : public Base
// send exit messages
for (actor_ptr& aptr : mlinks)
{
aptr->enqueue(message(mself, aptr, atom(":Exit"), reason));
aptr->enqueue(make_tuple(atom(":Exit"), mself, reason));
}
}
for (attachable_ptr& ptr : mattachables)
......@@ -228,7 +228,7 @@ class abstract_actor : public Base
}
if (reason != exit_reason::not_exited)
{
other->enqueue(message(this, other, atom(":Exit"), reason));
other->enqueue(make_tuple(atom(":Exit"), actor_ptr(this), reason));
}
return result;
}
......
......@@ -3,7 +3,7 @@
#include <memory>
#include "cppa/message.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/message_queue.hpp"
namespace cppa { namespace detail {
......@@ -12,7 +12,7 @@ template<class Super>
class abstract_message_queue : public Super
{
message m_last_dequeued;
any_tuple m_last_dequeued;
public:
......@@ -21,7 +21,7 @@ class abstract_message_queue : public Super
{
}
const message& dequeue() /*override*/
const any_tuple& dequeue() /*override*/
{
for (;;)
{
......@@ -56,7 +56,7 @@ class abstract_message_queue : public Super
}
}
bool try_dequeue(message& msg) /*override*/
bool try_dequeue(any_tuple& msg) /*override*/
{
for (;;)
{
......@@ -89,7 +89,7 @@ class abstract_message_queue : public Super
}
}
const message& last_dequeued() /*override*/
const any_tuple& last_dequeued() /*override*/
{
return m_last_dequeued;
}
......
#ifndef BLOCKING_MESSAGE_QUEUE_HPP
#define BLOCKING_MESSAGE_QUEUE_HPP
#include "cppa/message.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/message_queue.hpp"
#include "cppa/util/singly_linked_list.hpp"
#include "cppa/util/single_reader_queue.hpp"
......@@ -20,8 +20,8 @@ class blocking_message_queue_impl : public message_queue
struct queue_node
{
queue_node* next;
message msg;
queue_node(const message& from);
any_tuple msg;
queue_node(const any_tuple& from);
};
typedef util::single_reader_queue<queue_node> queue_type;
......@@ -36,7 +36,7 @@ class blocking_message_queue_impl : public message_queue
return m_queue;
}
virtual void enqueue(const message& msg) /*override*/;
virtual void enqueue(const any_tuple& msg) /*override*/;
protected:
......@@ -54,7 +54,7 @@ class blocking_message_queue_impl : public message_queue
// computes the next message, returns true if the computed message
// was not ignored (e.g. because it's an exit message with reason = normal)
bool dequeue_impl(message& storage);
bool dequeue_impl(any_tuple& storage);
bool dequeue_impl(invoke_rules&, queue_node_buffer&);
......
#ifndef MAILMAN_HPP
#define MAILMAN_HPP
#include "cppa/message.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/actor_proxy.hpp"
#include "cppa/process_information.hpp"
#include "cppa/detail/native_socket.hpp"
......@@ -12,9 +12,9 @@ namespace cppa { namespace detail {
struct mailman_send_job
{
process_information_ptr target_peer;
message original_message;
mailman_send_job(actor_proxy_ptr apptr, message msg);
mailman_send_job(process_information_ptr peer, message msg);
any_tuple original_message;
mailman_send_job(actor_proxy_ptr apptr, any_tuple msg);
mailman_send_job(process_information_ptr peer, any_tuple msg);
};
struct mailman_add_peer
......@@ -38,9 +38,9 @@ class mailman_job
kill_type
};
mailman_job(process_information_ptr piptr, const message& omsg);
mailman_job(process_information_ptr piptr, const any_tuple& omsg);
mailman_job(actor_proxy_ptr apptr, const message& omsg);
mailman_job(actor_proxy_ptr apptr, const any_tuple& omsg);
mailman_job(native_socket_t sockfd, const process_information_ptr& pinfo);
......
......@@ -3,7 +3,7 @@
#include <cstdint>
#include "cppa/message.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/message_queue.hpp"
#include "cppa/util/shared_spinlock.hpp"
#include "cppa/util/shared_lock_guard.hpp"
......@@ -34,8 +34,8 @@ class yielding_message_queue_impl : public message_queue
struct queue_node
{
queue_node* next;
message msg;
queue_node(const message& from);
any_tuple msg;
queue_node(const any_tuple& from);
};
bool m_has_pending_timeout_request;
......@@ -61,7 +61,7 @@ class yielding_message_queue_impl : public message_queue
ordinary_message
};
filter_result filter_msg(const message& msg);
filter_result filter_msg(const any_tuple& msg);
protected:
......@@ -83,7 +83,7 @@ class yielding_message_queue_impl : public message_queue
// conputes the next message, returns true if the computed message
// was not ignored (e.g. because it's an exit message with reason = normal)
bool dequeue_impl(message& storage);
bool dequeue_impl(any_tuple& storage);
bool dequeue_impl(invoke_rules&, queue_node_buffer&);
......@@ -95,7 +95,7 @@ class yielding_message_queue_impl : public message_queue
~yielding_message_queue_impl();
virtual void enqueue(const message& msg) /*override*/;
virtual void enqueue(const any_tuple& msg) /*override*/;
};
......
......@@ -74,12 +74,6 @@ class invoke_rules_base
invokable_list m_list;
invoke_rules_base() = default;
invoke_rules_base(invokable_list&& ilist);
invoke_rules_base(invoke_rules_base&& other);
public:
virtual ~invoke_rules_base();
......
......@@ -41,7 +41,7 @@ class local_actor : public actor
*
* Calls <code>mailbox().enqueue(msg)</code>.
*/
virtual void enqueue(const message& msg) /*override*/;
virtual void enqueue(const any_tuple& msg) /*override*/;
inline bool trap_exit() const;
......
#ifndef MESSAGE_HPP
#define MESSAGE_HPP
#include "cppa/actor.hpp"
#include "cppa/tuple.hpp"
#include "cppa/channel.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/tuple_view.hpp"
#include "cppa/ref_counted.hpp"
#include "cppa/intrusive_ptr.hpp"
#include "cppa/detail/channel.hpp"
namespace cppa {
struct msg_content : ref_counted
{
const actor_ptr sender;
const channel_ptr receiver;
const any_tuple data;
inline msg_content(const actor_ptr& s,
const channel_ptr& r,
const any_tuple& ut)
: ref_counted(), sender(s), receiver(r), data(ut)
{
}
inline msg_content(const actor_ptr& s,
const channel_ptr& r,
any_tuple&& ut)
: ref_counted(), sender(s), receiver(r), data(std::move(ut))
{
}
};
class message
{
public:
template<typename... Args>
message(const actor_ptr& from, const channel_ptr& to, const Args&... args);
message(const actor_ptr& from,
const channel_ptr& to,
const any_tuple& ut);
message(const actor_ptr& from,
const channel_ptr& to,
any_tuple&& ut);
message();
message& operator=(const message&) = default;
message& operator=(message&&) = default;
message(const message&) = default;
message(message&&) = default;
inline const actor_ptr& sender() const
{
return m_content->sender;
}
inline const channel_ptr& receiver() const
{
return m_content->receiver;
}
inline const any_tuple& content() const
{
return m_content->data;
}
bool empty() const;
private:
intrusive_ptr<msg_content> m_content;
};
template<typename... Args>
message::message(const actor_ptr& from, const channel_ptr& to, const Args&... args)
: m_content(new msg_content(from, to, make_tuple(args...)))//tuple<Args...>(args...)))
{
}
bool operator==(const message& lhs, const message& rhs);
inline bool operator!=(const message& lhs, const message& rhs)
{
return !(lhs == rhs);
}
} // namespace cppa
#endif // MESSAGE_HPP
#ifndef MESSAGE_QUEUE_HPP
#define MESSAGE_QUEUE_HPP
#include "cppa/message.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/ref_counted.hpp"
namespace cppa {
......@@ -19,7 +19,7 @@ class message_queue : public ref_counted
protected:
bool m_trap_exit;
message m_last_dequeued;
any_tuple m_last_dequeued;
public:
......@@ -32,14 +32,14 @@ class message_queue : public ref_counted
* @brief Enqueues a new element to the message queue.
* @param msg The new message.
*/
virtual void enqueue(const message& msg) = 0;
virtual void enqueue(const any_tuple& msg) = 0;
/**
* @brief Dequeues the oldest message (FIFO order) from the message queue.
* @returns The oldest message from the queue.
* @warning Call only from the owner of the queue.
*/
virtual const message& dequeue() = 0;
virtual const any_tuple& dequeue() = 0;
/**
* @brief Removes the first element from the queue that is matched
......@@ -60,7 +60,7 @@ class message_queue : public ref_counted
*
* @warning Call only from the owner of the queue.
*/
virtual bool try_dequeue(message&) = 0;
virtual bool try_dequeue(any_tuple&) = 0;
/**
*
......@@ -86,7 +86,7 @@ class message_queue : public ref_counted
* by a dequeue() or try_dequeue() member function call.
* @warning Call only from the owner of the queue.
*/
inline const message& last_dequeued() const;
inline const any_tuple& last_dequeued() const;
};
......@@ -104,7 +104,7 @@ inline void message_queue::trap_exit(bool value)
m_trap_exit = value;
}
inline const message& message_queue::last_dequeued() const
inline const any_tuple& message_queue::last_dequeued() const
{
return m_last_dequeued;
}
......
......@@ -8,7 +8,7 @@ namespace cppa {
/**
* @brief Dequeues the next message from the mailbox.
*/
inline const message& receive()
inline const any_tuple& receive()
{
return self()->mailbox().dequeue();
}
......@@ -59,7 +59,7 @@ void receive(invoke_rules& rules, Head&& head, Tail&&... tail)
* @returns @p true if a messages was dequeued;
* @p false if the mailbox is empty
*/
inline bool try_receive(message& msg)
inline bool try_receive(any_tuple& msg)
{
return self()->mailbox().try_dequeue(msg);
}
......
......@@ -91,7 +91,7 @@ class scheduler
{
static_assert(sizeof...(Data) > 0, "no message to send");
any_tuple tup = make_tuple(util::duration(rel_time), data...);
future_send_helper()->enqueue(message(from, to, tup));
future_send_helper()->enqueue(any_tuple(from, to, tup));
}
};
......
#include "cppa/atom.hpp"
#include "cppa/message.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/scheduler.hpp"
#include "cppa/actor_proxy.hpp"
#include "cppa/exit_reason.hpp"
......@@ -23,42 +23,40 @@ actor_proxy::actor_proxy(std::uint32_t mid, const process_information_ptr& pptr)
}
void actor_proxy::forward_message(const process_information_ptr& piptr,
const message& msg)
const any_tuple& msg)
{
detail::mailman_queue().push_back(new detail::mailman_job(piptr, msg));
}
void actor_proxy::enqueue(const message& msg)
void actor_proxy::enqueue(const any_tuple& msg)
{
const any_tuple& content = msg.content();
if ( content.size() > 0
&& content.utype_info_at(0) == typeid(atom_value))
if (msg.size() > 0 && msg.utype_info_at(0) == typeid(atom_value))
{
switch (to_int(*reinterpret_cast<const atom_value*>(content.at(0))))
if (msg.size() == 2 && msg.utype_info_at(1) == typeid(actor_ptr))
{
switch (to_int(msg.get_as<atom_value>(0)))
{
case to_int(atom(":Link")):
{
auto s = msg.sender();
auto s = msg.get_as<actor_ptr>(1);
link_to(s);
return;
}
case to_int(atom(":Unlink")):
{
auto s = msg.sender();
auto s = msg.get_as<actor_ptr>(1);
unlink_from(s);
return;
}
case to_int(atom(":KillProxy")):
{
if ( content.size() == 2
&& content.utype_info_at(1) == typeid(std::uint32_t))
{
const void* reason = content.at(1);
cleanup(*reinterpret_cast<const std::uint32_t*>(reason));
default: break;
}
return;
}
default: break;
else if ( msg.size() == 2
&& msg.get_as<atom_value>(0) == atom(":KillProxy")
&& msg.utype_info_at(1) == typeid(std::uint32_t))
{
cleanup(msg.get_as<std::uint32_t>(1));
return;
}
}
forward_message(parent_process_ptr(), msg);
......@@ -70,8 +68,7 @@ void actor_proxy::link_to(intrusive_ptr<actor>& other)
{
// causes remote actor to link to (proxy of) other
forward_message(parent_process_ptr(),
message(this, other, atom(":Link")));
//enqueue(message(this, other, atom(":Link")));
make_tuple(atom(":Link"), actor_ptr(this)));
}
}
......@@ -81,8 +78,7 @@ void actor_proxy::unlink_from(intrusive_ptr<actor>& other)
{
// causes remote actor to unlink from (proxy of) other
forward_message(parent_process_ptr(),
message(this, other, atom(":Unlink")));
//enqueue(message(this, other, atom(":Unlink")));
make_tuple(atom(":Unlink"), actor_ptr(this)));
}
}
......@@ -92,9 +88,8 @@ bool actor_proxy::establish_backlink(intrusive_ptr<actor>& other)
if (result)
{
forward_message(parent_process_ptr(),
message(this, other, atom(":Link")));
make_tuple(atom(":Link"), actor_ptr(this)));
}
//enqueue(message(to, this, atom(":Link")));
return result;
}
......@@ -104,9 +99,8 @@ bool actor_proxy::remove_backlink(intrusive_ptr<actor>& other)
if (result)
{
forward_message(parent_process_ptr(),
message(this, other, atom(":Unlink")));
make_tuple(atom(":Unlink"), actor_ptr(this)));
}
//enqueue(message(to, this, atom(":Unlink")));
return result;
}
......
#include "cppa/atom.hpp"
#include "cppa/message.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/detail/thread.hpp"
#include "cppa/detail/actor_proxy_cache.hpp"
......
......@@ -21,11 +21,11 @@ enum throw_on_exit_result
normal_exit_signal
};
throw_on_exit_result throw_on_exit(const message& msg)
throw_on_exit_result throw_on_exit(const any_tuple& msg)
{
if (match<atom_value, std::uint32_t>(msg.content(), nullptr, atom(":Exit")))
if (match<atom_value, actor_ptr, std::uint32_t>(msg, 0, atom(":Exit")))
{
auto reason = *reinterpret_cast<const std::uint32_t*>(msg.content().at(1));
auto reason = *reinterpret_cast<const std::uint32_t*>(msg.at(1));
if (reason != exit_reason::normal)
{
// throws
......@@ -43,17 +43,17 @@ throw_on_exit_result throw_on_exit(const message& msg)
namespace cppa { namespace detail {
blocking_message_queue_impl::queue_node::queue_node(const message& from)
blocking_message_queue_impl::queue_node::queue_node(const any_tuple& from)
: next(nullptr), msg(from)
{
}
void blocking_message_queue_impl::enqueue(const message& msg)
void blocking_message_queue_impl::enqueue(const any_tuple& msg)
{
m_queue.push_back(new queue_node(msg));
}
bool blocking_message_queue_impl::dequeue_impl(message& storage)
bool blocking_message_queue_impl::dequeue_impl(any_tuple& storage)
{
std::unique_ptr<queue_node> node(m_queue.pop());
if (!m_trap_exit)
......@@ -77,7 +77,7 @@ bool blocking_message_queue_impl::dq(std::unique_ptr<queue_node>& node,
{
return false;
}
std::unique_ptr<intermediate> imd(rules.get_intermediate(node->msg.content()));
std::unique_ptr<intermediate> imd(rules.get_intermediate(node->msg));
if (imd)
{
m_last_dequeued = node->msg;
......
#include "cppa/cppa.hpp"
#include "cppa/local_actor.hpp"
namespace {
......@@ -13,7 +14,8 @@ class observer : public cppa::attachable
void detach(std::uint32_t reason)
{
cppa::send(m_client, cppa::atom(":Down"), reason);
using namespace cppa;
send(m_client, atom(":Down"), actor_ptr(self()), reason);
}
bool matches(const cppa::attachable::token& match_token)
......@@ -34,14 +36,14 @@ namespace cppa {
local_actor* operator<<(local_actor* whom, const any_tuple& what)
{
if (whom) whom->enqueue(message(self(), whom, what));
if (whom) whom->enqueue(what);
return whom;
}
// matches self() << make_tuple(...)
local_actor* operator<<(local_actor* whom, any_tuple&& what)
{
if (whom) whom->enqueue(message(self(), whom, std::move(what)));
if (whom) whom->enqueue(std::move(what));
return whom;
}
......
......@@ -35,7 +35,7 @@ class local_group : public group
public:
virtual void enqueue(const message& msg)
virtual void enqueue(const any_tuple& msg)
{
shared_guard guard(m_shared_mtx);
for (auto i = m_subscribers.begin(); i != m_subscribers.end(); ++i)
......
......@@ -3,7 +3,7 @@
#include <boost/thread.hpp>
#include "cppa/local_actor.hpp"
#include "cppa/message.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/scheduler.hpp"
#include "cppa/detail/converted_thread_context.hpp"
......@@ -29,7 +29,7 @@ boost::thread_specific_ptr<cppa::local_actor> t_this_context(cleanup_fun);
namespace cppa {
void local_actor::enqueue(const message& msg)
void local_actor::enqueue(const any_tuple& msg)
{
mailbox().enqueue(msg);
}
......
......@@ -12,12 +12,12 @@
// implementation of mailman.hpp
namespace cppa { namespace detail {
mailman_send_job::mailman_send_job(actor_proxy_ptr apptr, message msg)
mailman_send_job::mailman_send_job(actor_proxy_ptr apptr, any_tuple msg)
: target_peer(apptr->parent_process_ptr()), original_message(msg)
{
}
mailman_send_job::mailman_send_job(process_information_ptr peer, message msg)
mailman_send_job::mailman_send_job(process_information_ptr peer, any_tuple msg)
: target_peer(peer), original_message(msg)
{
}
......@@ -32,13 +32,13 @@ mailman_job::mailman_job(job_type jt) : next(0), m_type(jt)
{
}
mailman_job::mailman_job(process_information_ptr piptr, const message& omsg)
mailman_job::mailman_job(process_information_ptr piptr, const any_tuple& omsg)
: next(0), m_type(send_job_type)
{
new (&m_send_job) mailman_send_job(piptr, omsg);
}
mailman_job::mailman_job(actor_proxy_ptr apptr, const message& omsg)
mailman_job::mailman_job(actor_proxy_ptr apptr, const any_tuple& omsg)
: next(0), m_type(send_job_type)
{
new (&m_send_job) mailman_send_job(apptr, omsg);
......@@ -103,7 +103,7 @@ void mailman_loop()
if (job->is_send_job())
{
mailman_send_job& sjob = job->send_job();
const message& out_msg = sjob.original_message;
const any_tuple& out_msg = sjob.original_message;
// forward message to receiver peer
auto peer_element = peers.find(*(sjob.target_peer));
if (peer_element != peers.end())
......
#include "cppa/message.hpp"
#include "cppa/detail/singleton_manager.hpp"
namespace cppa {
message::message(const actor_ptr& from,
const channel_ptr& to,
const any_tuple& ut)
: m_content(new msg_content(from, to, ut))
{
}
message::message(const actor_ptr& from,
const channel_ptr& to,
any_tuple&& ut)
: m_content(new msg_content(from, to, std::move(ut)))
{
}
message::message() : m_content(detail::singleton_manager::get_message_dummy())
{
}
bool message::empty() const
{
return m_content.get() == detail::singleton_manager::get_message_dummy();
}
bool operator==(const message& lhs, const message& rhs)
{
return lhs.sender() == rhs.sender()
&& lhs.receiver() == rhs.receiver()
&& lhs.content().vals()->equal_to(*(rhs.content().vals()));
}
} // namespace cppa
......@@ -4,7 +4,7 @@
#include <iostream>
#include "cppa/message.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/scheduler.hpp"
#include "cppa/attachable.hpp"
......
......@@ -189,14 +189,14 @@ class po_peer : public post_office_worker
, m_state(wait_for_msg_size)
, m_peer(std::move(from.peer))
, m_observer(std::move(from.attachable_ptr))
, m_meta_msg(uniform_typeid<message>())
, m_meta_msg(uniform_typeid<any_tuple>())
{
}
explicit po_peer(native_socket_t sockfd, native_socket_t parent_socket)
: super(sockfd, parent_socket)
, m_state(wait_for_process_info)
, m_meta_msg(uniform_typeid<message>())
, m_meta_msg(uniform_typeid<any_tuple>())
{
m_rdbuf.reset( sizeof(std::uint32_t)
+ process_information::node_id_size);
......@@ -209,7 +209,7 @@ class po_peer : public post_office_worker
, m_observer(std::move(other.m_observer))
, m_rdbuf(std::move(other.m_rdbuf))
, m_children(std::move(other.m_children))
, m_meta_msg(uniform_typeid<message>())
, m_meta_msg(uniform_typeid<any_tuple>())
{
}
......@@ -224,7 +224,7 @@ class po_peer : public post_office_worker
{
for (actor_proxy_ptr& pptr : m_children)
{
pptr->enqueue(message(nullptr, nullptr, atom(":KillProxy"),
pptr->enqueue(make_tuple(atom(":KillProxy"),
exit_reason::remote_link_unreachable));
}
}
......@@ -296,7 +296,7 @@ class po_peer : public post_office_worker
// wait for new data
break;
}
message msg;
any_tuple msg;
binary_deserializer bd(m_rdbuf.data(), m_rdbuf.size());
try
{
......@@ -308,13 +308,12 @@ class po_peer : public post_office_worker
DEBUG(to_uniform_name(typeid(e)) << ": " << e.what());
return false;
}
auto& content = msg.content();
if ( content.size() == 1
&& content.utype_info_at(0) == typeid(atom_value)
&& *reinterpret_cast<const atom_value*>(content.at(0))
== atom(":Monitor"))
if ( msg.size() == 2
&& msg.utype_info_at(0) == typeid(atom_value)
&& msg.get_as<atom_value>(0) == atom(":Monitor")
&& msg.utype_info_at(1) == typeid(actor_ptr))
{
actor_ptr sender = msg.sender();
actor_ptr sender = msg.get_as<actor_ptr>(1);
if (sender->parent_process() == *process_information::get())
{
//cout << pinfo << " ':Monitor'; actor id = "
......@@ -323,8 +322,8 @@ class po_peer : public post_office_worker
// this message was send from a proxy
sender->attach_functor([=](std::uint32_t reason)
{
message msg(sender, sender,
atom(":KillProxy"), reason);
any_tuple msg = make_tuple(atom(":KillProxy"),
reason);
auto mjob = new detail::mailman_job(m_peer, msg);
detail::mailman_queue().push_back(mjob);
});
......@@ -471,7 +470,7 @@ void post_office_loop(int pipe_read_handle, int pipe_write_handle)
// initialize proxy cache
get_actor_proxy_cache().set_callback([&](actor_proxy_ptr& pptr)
{
pptr->enqueue(message(pptr, nullptr, atom(":Monitor")));
pptr->enqueue(make_tuple(atom(":Monitor"), pptr));
if (selected_peer == nullptr)
{
throw std::logic_error("selected_peer == nullptr");
......
......@@ -47,7 +47,7 @@ struct scheduler_helper
{
{
any_tuple content = make_tuple(atom(":_DIE"));
m_worker->enqueue(message(m_worker, m_worker, content));
m_worker->enqueue(any_tuple(m_worker, m_worker, content));
}
m_thread.join();
}
......@@ -110,7 +110,7 @@ void scheduler_helper::time_emitter(scheduler_helper::ptr_type m_self)
{
auto& whom = (it->second).first;
auto& what = (it->second).second;
whom->enqueue(message(whom, whom, what));
whom->enqueue(any_tuple(whom, whom, what));
messages.erase(it);
it = messages.begin();
}
......
#include <atomic>
#include <iostream>
#include "cppa/message.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/local_actor.hpp"
#include "cppa/scheduler.hpp"
#include "cppa/exception.hpp"
......
......@@ -10,7 +10,7 @@
#include "cppa/actor.hpp"
#include "cppa/group.hpp"
#include "cppa/channel.hpp"
#include "cppa/message.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/any_type.hpp"
#include "cppa/any_tuple.hpp"
......@@ -96,7 +96,7 @@ std::string to_uniform_name_impl(Iterator begin, Iterator end,
{ demangled<cppa::actor_ptr>(), "@actor" },
{ demangled<cppa::group_ptr>(), "@group" },
{ demangled<cppa::channel_ptr>(), "@channel" },
{ demangled<cppa::message>(), "@msg" }
{ demangled<cppa::any_tuple>(), "@msg" }
};
// check if we could find the whole string in our lookup map
......
......@@ -15,7 +15,7 @@
#include "cppa/atom.hpp"
#include "cppa/actor.hpp"
#include "cppa/object.hpp"
#include "cppa/message.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/announce.hpp"
#include "cppa/any_type.hpp"
#include "cppa/any_tuple.hpp"
......@@ -441,7 +441,7 @@ class any_tuple_tinfo : public util::abstract_uniform_type_info<any_tuple>
};
class message_tinfo : public util::abstract_uniform_type_info<message>
class message_tinfo : public util::abstract_uniform_type_info<any_tuple>
{
std::string any_tuple_name;
......@@ -453,7 +453,7 @@ class message_tinfo : public util::abstract_uniform_type_info<message>
virtual void serialize(const void* instance, serializer* sink) const
{
const message& msg = *reinterpret_cast<const message*>(instance);
const any_tuple& msg = *reinterpret_cast<const any_tuple*>(instance);
const any_tuple& data = msg.content();
sink->begin_object(name());
actor_ptr_tinfo::s_serialize(msg.sender().get(), sink, actor_ptr_name);
......@@ -488,7 +488,7 @@ class message_tinfo : public util::abstract_uniform_type_info<message>
//uniform_typeid<channel_ptr>()->deserialize(&receiver, source);
//uniform_typeid<any_tuple>()->deserialize(&content, source);
source->end_object();
*reinterpret_cast<message*>(instance) = message(sender,
*reinterpret_cast<any_tuple*>(instance) = any_tuple(sender,
receiver,
content);
}
......@@ -656,7 +656,7 @@ class uniform_type_info_map_helper
insert(d, new actor_ptr_tinfo, { raw_name<actor_ptr>() });
insert(d, new group_ptr_tinfo, { raw_name<actor_ptr>() });
insert(d, new channel_ptr_tinfo, { raw_name<channel_ptr>() });
insert(d, new message_tinfo, { raw_name<message>() });
insert(d, new message_tinfo, { raw_name<any_tuple>() });
insert(d, new atom_value_tinfo, { raw_name<atom_value>() });
insert<float>(d);
insert<cppa::util::void_type>(d);
......
......@@ -29,7 +29,7 @@ typedef cppa::util::shared_lock_guard<cppa::util::shared_spinlock> shared_guard;
namespace cppa { namespace detail {
yielding_message_queue_impl::queue_node::queue_node(const message& from)
yielding_message_queue_impl::queue_node::queue_node(const any_tuple& from)
: next(0), msg(from)
{
}
......@@ -53,10 +53,10 @@ enum filter_result
};
yielding_message_queue_impl::filter_result
yielding_message_queue_impl::filter_msg(const message& msg)
yielding_message_queue_impl::filter_msg(const any_tuple& msg)
{
if ( m_trap_exit == false
&& match<atom_value, std::uint32_t>(msg.content(),
&& match<atom_value, actor_ptr, std::uint32_t>(msg,
nullptr,
atom(":Exit")))
{
......@@ -79,7 +79,7 @@ yielding_message_queue_impl::filter_msg(const message& msg)
return ordinary_message;
}
void yielding_message_queue_impl::enqueue(const message& msg)
void yielding_message_queue_impl::enqueue(const any_tuple& msg)
{
if (m_queue._push_back(new queue_node(msg)))
{
......@@ -131,7 +131,7 @@ void yielding_message_queue_impl::yield_until_not_empty()
}
}
bool yielding_message_queue_impl::dequeue_impl(message& storage)
bool yielding_message_queue_impl::dequeue_impl(any_tuple& storage)
{
yield_until_not_empty();
std::unique_ptr<queue_node> node(m_queue.pop());
......
......@@ -56,7 +56,7 @@ size_t test__atom()
}
);
CPPA_CHECK(matched_pattern[0] && matched_pattern[1] && matched_pattern[2]);
message msg = receive();
any_tuple msg = receive();
CPPA_CHECK((match<atom_value, atom_value, atom_value, float>(msg.content(), nullptr, atom("b"), atom("a"), atom("c"))));
CPPA_CHECK(try_receive(msg) == false);
return CPPA_TEST_RESULT;
......
......@@ -48,7 +48,7 @@ size_t test__local_group()
)
.until([&result]() { return result == 10; });
await_all_others_done();
message tmp;
any_tuple tmp;
CPPA_CHECK_EQUAL(try_receive(tmp), false);
return CPPA_TEST_RESULT;
}
......@@ -26,7 +26,7 @@
#include "cppa/match.hpp"
#include "cppa/tuple.hpp"
#include "cppa/message.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/announce.hpp"
#include "cppa/get_view.hpp"
#include "cppa/any_tuple.hpp"
......@@ -157,7 +157,7 @@ size_t test__serialization()
}
{
message msg1(0, 0, 42, std::string("Hello \"World\"!"));
any_tuple msg1(0, 0, 42, std::string("Hello \"World\"!"));
CPPA_CHECK_EQUAL(msg1str, to_string(msg1));
binary_serializer bs;
bs << msg1;
......@@ -166,10 +166,10 @@ size_t test__serialization()
bd >> obj1;
object obj2 = from_string(to_string(msg1));
CPPA_CHECK_EQUAL(obj1, obj2);
if (obj1.type() == typeid(message) && obj2.type() == obj1.type())
if (obj1.type() == typeid(any_tuple) && obj2.type() == obj1.type())
{
auto& content1 = get<message>(obj1).content();
auto& content2 = get<message>(obj2).content();
auto& content1 = get<any_tuple>(obj1).content();
auto& content2 = get<any_tuple>(obj2).content();
auto cview1 = get_view<decltype(42), std::string>(content1);
auto cview2 = get_view<decltype(42), std::string>(content2);
CPPA_CHECK_EQUAL(cview1.size(), 2);
......
......@@ -24,9 +24,9 @@ void testee1()
(
others() >> []()
{
message msg = last_received();
any_tuple msg = last_received();
actor_ptr sender = msg.sender();
sender->enqueue(message(self(), sender, msg.content()));
sender->enqueue(any_tuple(self(), sender, msg.content()));
},
after(std::chrono::milliseconds(10)) >> []()
{
......@@ -87,16 +87,18 @@ size_t test__spawn()
// wait for :Down and :Exit messages of pong
receive_while([&i]() { return ++i <= 4; })
(
on<atom(":Exit"), std::uint32_t>() >> [&](std::uint32_t reason)
on<atom(":Exit"), actor_ptr, std::uint32_t>() >> [&](const actor_ptr& who,
std::uint32_t reason)
{
CPPA_CHECK_EQUAL(reason, exit_reason::user_defined);
CPPA_CHECK_EQUAL(last_received().sender(), pong_actor);
CPPA_CHECK_EQUAL(who, pong_actor);
flags |= 0x01;
},
on<atom(":Down"), std::uint32_t>() >> [&](std::uint32_t reason)
on<atom(":Down"), actor_ptr, std::uint32_t>() >> [&](const actor_ptr& who,
std::uint32_t reason)
{
CPPA_CHECK_EQUAL(reason, exit_reason::user_defined);
if (last_received().sender() == pong_actor)
if (who == pong_actor)
{
flags |= 0x02;
}
......@@ -125,7 +127,7 @@ size_t test__spawn()
await_all_others_done();
CPPA_CHECK_EQUAL(flags, 0x0F);
// mailbox has to be empty
message msg;
any_tuple msg;
while (try_receive(msg))
{
report_unexpected();
......
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