Commit b1bb7832 authored by Dominik Charousset's avatar Dominik Charousset

removed class `network::addressed_message`

rather than wrapping a tuple, we use `message_header` to prefix any tuple
in network communication; this enables a more flexible handling and allows
for receiver-dependent deserialization and to forward messages without
deserializing it
parent 8610c520
......@@ -89,7 +89,6 @@ set(LIBCPPA_SRC
src/actor_count.cpp
src/actor_proxy.cpp
src/actor_registry.cpp
src/addressed_message.cpp
src/any_tuple.cpp
src/atom.cpp
src/attachable.cpp
......@@ -122,6 +121,7 @@ set(LIBCPPA_SRC
src/ipv4_io_stream.cpp
src/local_actor.cpp
src/logging.cpp
src/message_header.cpp
src/middleman.cpp
src/object.cpp
src/object_array.cpp
......
......@@ -40,7 +40,6 @@ cppa/detail/abstract_tuple.hpp
cppa/detail/actor_count.hpp
cppa/actor_addressing.hpp
cppa/detail/actor_registry.hpp
cppa/network/addressed_message.hpp
cppa/detail/atom_val.hpp
cppa/detail/boxed.hpp
cppa/detail/channel.hpp
......@@ -174,7 +173,6 @@ src/actor.cpp
src/actor_count.cpp
src/actor_proxy.cpp
src/actor_registry.cpp
src/addressed_message.cpp
src/any_tuple.cpp
src/atom.cpp
src/attachable.cpp
......@@ -295,3 +293,5 @@ cppa/detail/singleton_mixin.hpp
cppa/response_handle.hpp
src/response_handle.cpp
cppa/wildcard_position.hpp
cppa/network/message_header.hpp
src/message_header.cpp
......@@ -35,6 +35,8 @@
namespace cppa {
struct invalid_message_id_flag { constexpr invalid_message_id_flag() { } };
/**
* @brief
* @note Asynchronous messages always have an invalid message id.
......@@ -99,9 +101,13 @@ class message_id_t {
return result;
}
static constexpr invalid_message_id_flag invalid = invalid_message_id_flag();
inline message_id_t(invalid_message_id_flag) : m_value(0) { }
private:
explicit message_id_t(std::uint64_t value) : m_value(value) { }
explicit constexpr message_id_t(std::uint64_t value) : m_value(value) { }
std::uint64_t m_value;
......
......@@ -68,7 +68,7 @@ class default_peer : public continuable_reader, public continuable_writer {
continuable_writer* as_writer();
void enqueue(const addressed_message& msg);
void enqueue(const message_header& hdr, const any_tuple& msg);
inline bool erase_on_last_proxy_exited() const {
return m_erase_on_last_proxy_exited;
......@@ -100,9 +100,11 @@ class default_peer : public continuable_reader, public continuable_writer {
output_stream_ptr m_out;
read_state m_state;
process_information_ptr m_node;
const uniform_type_info* m_meta_msg;
bool m_has_unwritten_data;
const uniform_type_info* m_meta_hdr;
const uniform_type_info* m_meta_msg;
util::buffer m_rd_buf;
util::buffer m_wr_buf;
......@@ -120,10 +122,10 @@ class default_peer : public continuable_reader, public continuable_writer {
void unlink(const actor_ptr& sender, const actor_ptr& ptr);
void deliver(const addressed_message& msg);
void deliver(const message_header& hdr, any_tuple msg);
inline void enqueue(const any_tuple& msg) {
enqueue(addressed_message(nullptr, nullptr, msg));
enqueue({nullptr, nullptr}, msg);
}
template<typename Arg0, typename Arg1, typename... Args>
......
......@@ -28,94 +28,44 @@
\******************************************************************************/
#ifndef CPPA_ADDRESSED_MESSAGE_HPP
#define CPPA_ADDRESSED_MESSAGE_HPP
#ifndef CPPA_MESSAGE_HEADER_HPP
#define CPPA_MESSAGE_HEADER_HPP
#include "cppa/actor.hpp"
#include "cppa/channel.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/cow_tuple.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/ref_counted.hpp"
#include "cppa/intrusive_ptr.hpp"
#include "cppa/message_id.hpp"
namespace cppa { namespace network {
/**
* @brief Encapsulates a message along with sender and receiver information
* as well as its synchronous message id.
* @brief Encapsulates information about sender, receiver and (synchronous)
* message ID of a message. The message itself is usually an any_tuple.
*/
class addressed_message {
class message_header {
public:
addressed_message(actor_ptr from, channel_ptr to,
any_tuple ut, message_id_t id = message_id_t());
actor_ptr sender;
actor_ptr receiver;
message_id_t id;
addressed_message() = default;
addressed_message(addressed_message&&) = default;
addressed_message(const addressed_message&) = default;
addressed_message& operator=(addressed_message&&) = default;
addressed_message& operator=(const addressed_message&) = default;
message_header();
inline actor_ptr& sender() {
return m_sender;
}
inline const actor_ptr& sender() const {
return m_sender;
}
inline channel_ptr& receiver() {
return m_receiver;
}
inline const channel_ptr& receiver() const {
return m_receiver;
}
inline any_tuple& content() {
return m_content;
}
inline const any_tuple& content() const {
return m_content;
}
inline message_id_t id() const {
return m_msg_id;
}
inline void id(message_id_t value) {
m_msg_id = value;
}
inline bool empty() const {
return m_content.empty();
}
private:
actor_ptr m_sender;
channel_ptr m_receiver;
message_id_t m_msg_id;
any_tuple m_content;
message_header(const actor_ptr& sender,
const actor_ptr& receiver,
message_id_t id = message_id_t::invalid);
};
/**
* @relates addressed_message
*/
bool operator==(const addressed_message& lhs, const addressed_message& rhs);
inline bool operator==(const message_header& lhs, const message_header& rhs) {
return lhs.sender == rhs.sender
&& lhs.receiver == rhs.receiver
&& lhs.id == rhs.id;
}
/**
* @relates addressed_message
*/
inline bool operator!=(const addressed_message& lhs,
const addressed_message& rhs) {
inline bool operator!=(const message_header& lhs, const message_header& rhs) {
return !(lhs == rhs);
}
} } // namespace cppa::network
#endif // CPPA_ADDRESSED_MESSAGE_HPP
#endif // CPPA_MESSAGE_HEADER_HPP
......@@ -43,8 +43,7 @@
#include "cppa/intrusive_ptr.hpp"
#include "cppa/uniform_type_info.hpp"
#include "cppa/process_information.hpp"
#include "cppa/network/addressed_message.hpp"
#include "cppa/network/message_header.hpp"
namespace std { class exception; }
......@@ -65,7 +64,7 @@ inline std::string to_string(const any_tuple& what) {
return detail::to_string_impl(what);
}
inline std::string to_string(const network::addressed_message& what) {
inline std::string to_string(const network::message_header& what) {
return detail::to_string_impl(what);
}
......
......@@ -43,7 +43,7 @@
#include "cppa/process_information.hpp"
namespace cppa { class any_tuple; }
namespace cppa { namespace network { class addressed_message; } }
namespace cppa { namespace network { class message_header; } }
namespace cppa { namespace util {
......@@ -71,7 +71,7 @@ template<>
struct is_builtin<any_tuple> : std::true_type { };
template<>
struct is_builtin<network::addressed_message> : std::true_type { };
struct is_builtin<network::message_header> : std::true_type { };
template<>
struct is_builtin<actor_ptr> : std::true_type { };
......
......@@ -102,7 +102,7 @@ decorated_names_map::decorated_names_map() {
{ demangled<char32_t>(), mapped_int_name<char32_t>() },
{ "cppa::atom_value", "@atom" },
{ "cppa::any_tuple", "@<>" },
{ "cppa::network::addressed_message", "@msg" },
{ "cppa::network::message_header", "@hdr" },
{ "cppa::intrusive_ptr<cppa::actor>", "@actor" },
{ "cppa::intrusive_ptr<cppa::group>" ,"@group" },
{ "cppa::intrusive_ptr<cppa::channel>", "@channel" },
......
......@@ -146,7 +146,7 @@ void default_actor_addressing::put(const process_information& node,
auto p = m_parent->get_peer(node);
CPPA_LOG_ERROR_IF(!p, "put a proxy for an unknown peer");
if (p) {
p->enqueue(addressed_message(nullptr, nullptr, make_any_tuple(atom("MONITOR"), process_information::get(), aid)));
p->enqueue({nullptr, nullptr}, make_any_tuple(atom("MONITOR"), process_information::get(), aid));
}
}
else {
......
......@@ -78,7 +78,7 @@ void default_actor_proxy::forward_msg(const actor_ptr& sender, any_tuple msg, me
<< ", proto = " << to_string(proto->identifier()));
*/
auto p = proto->get_peer(*node);
if (p) p->enqueue(addressed_message(sender, receiver, msg, mid));
if (p) p->enqueue({sender, receiver, mid}, msg);
});
}
......
......@@ -46,8 +46,9 @@
#include "cppa/detail/singleton_manager.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/network/default_protocol.hpp"
#include "cppa/network/default_peer.hpp"
#include "cppa/network/message_header.hpp"
#include "cppa/network/default_protocol.hpp"
using namespace std;
......@@ -61,7 +62,6 @@ default_peer::default_peer(default_protocol* parent,
, m_parent(parent), m_in(in), m_out(out)
, m_state((peer_ptr) ? wait_for_msg_size : wait_for_process_info)
, m_node(peer_ptr)
, m_meta_msg(uniform_typeid<addressed_message>())
, m_has_unwritten_data(false) {
m_rd_buf.reset(m_state == wait_for_process_info
? sizeof(uint32_t) + process_information::node_id_size
......@@ -69,6 +69,8 @@ default_peer::default_peer(default_protocol* parent,
// state == wait_for_msg_size iff peer was created using remote_peer()
// in this case, this peer must be erased if no proxy of it remains
m_erase_on_last_proxy_exited = m_state == wait_for_msg_size;
m_meta_hdr = uniform_typeid<message_header>();
m_meta_msg = uniform_typeid<any_tuple>();
}
default_peer::~default_peer() {
......@@ -132,37 +134,40 @@ continue_reading_result default_peer::continue_reading() {
}
case read_message: {
//DEBUG("peer_connection::continue_reading: read_message");
addressed_message msg;
message_header hdr;
any_tuple msg;
binary_deserializer bd(m_rd_buf.data(), m_rd_buf.size(),
m_parent->addressing());
try { m_meta_msg->deserialize(&msg, &bd); }
try {
m_meta_hdr->deserialize(&hdr, &bd);
m_meta_msg->deserialize(&msg, &bd);
}
catch (exception& e) {
CPPA_LOG_ERROR("exception during read_message: "
<< detail::demangle(typeid(e))
<< ", what(): " << e.what());
return read_failure;
}
auto& content = msg.content();
CPPA_LOG_DEBUG("deserialized: " << to_string(msg));
//DEBUG("<-- " << to_string(msg));
match(content) (
match(msg) (
// monitor messages are sent automatically whenever
// actor_proxy_cache creates a new proxy
// note: aid is the *original* actor id
on(atom("MONITOR"), arg_match) >> [&](const process_information_ptr& node, actor_id aid) {
monitor(msg.sender(), node, aid);
monitor(hdr.sender, node, aid);
},
on(atom("KILL_PROXY"), arg_match) >> [&](const process_information_ptr& node, actor_id aid, std::uint32_t reason) {
kill_proxy(msg.sender(), node, aid, reason);
kill_proxy(hdr.sender, node, aid, reason);
},
on(atom("LINK"), arg_match) >> [&](const actor_ptr& ptr) {
link(msg.sender(), ptr);
link(hdr.sender, ptr);
},
on(atom("UNLINK"), arg_match) >> [&](const actor_ptr& ptr) {
unlink(msg.sender(), ptr);
unlink(hdr.sender, ptr);
},
others() >> [&] {
deliver(msg);
deliver(hdr, move(msg));
}
);
m_rd_buf.reset(sizeof(uint32_t));
......@@ -250,36 +255,23 @@ void default_peer::kill_proxy(const actor_ptr& sender,
}
}
void default_peer::deliver(const addressed_message& msg) {
void default_peer::deliver(const message_header& hdr, any_tuple msg) {
CPPA_LOG_TRACE("");
auto receiver = msg.receiver().get();
auto receiver = hdr.receiver.get();
if (receiver) {
if (msg.id().valid()) {
auto ra = dynamic_cast<actor*>(receiver);
if (ra) {
CPPA_LOG_DEBUG("sync message for actor "
<< ra->id());
ra->sync_enqueue(
msg.sender().get(),
msg.id(),
move(msg.content()));
}
else{
CPPA_LOG_ERROR("sync mesage to non-actor");
}
if (hdr.id.valid()) {
CPPA_LOG_DEBUG("sync message for actor " << receiver->id());
receiver->sync_enqueue(hdr.sender.get(), hdr.id, move(msg));
}
else {
CPPA_LOG_DEBUG("async message with "
<< (msg.sender() ? "" : "in")
<< "valid sender");
receiver->enqueue(
msg.sender().get(),
move(msg.content()));
receiver->enqueue(hdr.sender.get(), move(msg));
}
}
else {
CPPA_LOG_ERROR("received message with "
"invalid receiver");
CPPA_LOG_ERROR("received message with invalid receiver");
}
}
......@@ -344,13 +336,13 @@ continuable_writer* default_peer::as_writer() {
return this;
}
void default_peer::enqueue(const addressed_message& msg) {
void default_peer::enqueue(const message_header& hdr, const any_tuple& msg) {
CPPA_LOG_TRACE("");
binary_serializer bs(&m_wr_buf, m_parent->addressing());
uint32_t size = 0;
auto before = m_wr_buf.size();
m_wr_buf.write(sizeof(uint32_t), &size, util::grow_if_needed);
try { bs << msg; }
try { bs << hdr << msg; }
catch (exception& e) {
CPPA_LOG_ERROR(to_verbose_string(e));
cerr << "*** exception in default_peer::enqueue; "
......
......@@ -44,7 +44,7 @@
#include "cppa/network/middleman.hpp"
#include "cppa/detail/types_array.hpp"
#include "cppa/detail/group_manager.hpp"
#include "cppa/network/addressed_message.hpp"
#include "cppa/network/message_header.hpp"
#include "cppa/util/shared_spinlock.hpp"
#include "cppa/util/shared_lock_guard.hpp"
......
......@@ -28,23 +28,15 @@
\******************************************************************************/
#include "cppa/network/addressed_message.hpp"
#include "cppa/detail/singleton_manager.hpp"
#include "cppa/network/message_header.hpp"
namespace cppa { namespace network {
addressed_message::addressed_message(actor_ptr from,
channel_ptr to,
any_tuple ut,
message_id_t id )
: m_sender(std::move(from)), m_receiver(std::move(to))
, m_msg_id(id), m_content(std::move(ut)) { }
message_header::message_header(const actor_ptr& s,
const actor_ptr& r,
message_id_t mid )
: sender(s), receiver(r), id(mid) { }
bool operator==(const addressed_message& lhs, const addressed_message& rhs) {
return lhs.sender() == rhs.sender()
&& lhs.receiver() == rhs.receiver()
&& lhs.id() == rhs.id()
&& lhs.content() == rhs.content();
}
message_header::message_header() : sender(nullptr), receiver(nullptr), id() { }
} } // namespace cppa
} } // namespace cppa::network
......@@ -56,7 +56,6 @@
#include "cppa/network/input_stream.hpp"
#include "cppa/network/output_stream.hpp"
#include "cppa/network/default_protocol.hpp"
#include "cppa/network/addressed_message.hpp"
#include "cppa/network/middleman_event_handler_base.hpp"
#include "cppa/detail/fd_util.hpp"
......
......@@ -47,7 +47,7 @@
#include "cppa/detail/demangle.hpp"
#include "cppa/detail/to_uniform_name.hpp"
#include "cppa/network/addressed_message.hpp"
#include "cppa/network/message_header.hpp"
#include "cppa/detail/singleton_manager.hpp"
#include "cppa/detail/decorated_names_map.hpp"
......
......@@ -62,7 +62,7 @@
#include "cppa/detail/uniform_type_info_map.hpp"
#include "cppa/detail/default_uniform_type_info_impl.hpp"
#include "cppa/network/addressed_message.hpp"
#include "cppa/network/message_header.hpp"
namespace cppa { namespace detail {
......@@ -387,28 +387,20 @@ class any_tuple_tinfo : public util::abstract_uniform_type_info<any_tuple> {
};
class addr_msg_tinfo : public util::abstract_uniform_type_info<network::addressed_message> {
class msg_hdr_tinfo : public util::abstract_uniform_type_info<network::message_header> {
string any_tuple_name;
string actor_ptr_name;
string group_ptr_name;
string channel_ptr_name;
public:
virtual void serialize(const void* instance, serializer* sink) const {
CPPA_LOG_TRACE("");
auto& msg = *reinterpret_cast<const network::addressed_message*>(instance);
auto& data = msg.content();
auto& hdr = *reinterpret_cast<const network::message_header*>(instance);
sink->begin_object(name());
actor_ptr_tinfo::s_serialize(msg.sender(), sink, actor_ptr_name);
channel_ptr_tinfo::s_serialize(msg.receiver(),
sink,
channel_ptr_name,
actor_ptr_name,
group_ptr_name);
sink->write_value(msg.id().integer_value());
any_tuple_tinfo::s_serialize(data, sink, any_tuple_name);
actor_ptr_tinfo::s_serialize(hdr.sender, sink, actor_ptr_name);
actor_ptr_tinfo::s_serialize(hdr.receiver, sink, actor_ptr_name);
sink->write_value(hdr.id.integer_value());
sink->end_object();
}
......@@ -416,23 +408,17 @@ class addr_msg_tinfo : public util::abstract_uniform_type_info<network::addresse
CPPA_LOG_TRACE("");
assert_type_name(source);
source->begin_object(name());
auto& msg = *reinterpret_cast<network::addressed_message*>(instance);
actor_ptr_tinfo::s_deserialize(msg.sender(), source, actor_ptr_name);
channel_ptr_tinfo::s_deserialize(msg.receiver(),
source,
channel_ptr_name,
actor_ptr_name,
group_ptr_name);
auto& msg = *reinterpret_cast<network::message_header*>(instance);
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_value(pt_uint64);
msg.id(message_id_t::from_integer_value(get<pt_uint64>(msg_id)));
any_tuple_tinfo::s_deserialize(msg.content(), source, any_tuple_name);
msg.id = message_id_t::from_integer_value(get<pt_uint64>(msg_id));
source->end_object();
}
addr_msg_tinfo() : any_tuple_name(to_uniform_name<any_tuple>())
, actor_ptr_name(to_uniform_name<actor_ptr>())
, group_ptr_name(to_uniform_name<group_ptr>())
, channel_ptr_name(to_uniform_name<channel_ptr>()) { }
msg_hdr_tinfo() : any_tuple_name(to_uniform_name<any_tuple>())
, actor_ptr_name(to_uniform_name<actor_ptr>()) { }
};
......@@ -649,7 +635,7 @@ uniform_type_info_map::uniform_type_info_map() {
insert({raw_name<group_ptr>()}, new group_ptr_tinfo);
insert({raw_name<channel_ptr>()}, new channel_ptr_tinfo);
insert({raw_name<atom_value>()}, new atom_value_tinfo);
insert({raw_name<network::addressed_message>()}, new addr_msg_tinfo);
insert({raw_name<network::message_header>()}, new msg_hdr_tinfo);
insert({raw_name<util::void_type>()}, new void_type_tinfo);
insert({raw_name<process_information_ptr>()}, new process_info_ptr_tinfo);
insert({raw_name<map<string,string>>()}, new default_uniform_type_info_impl<map<string,string>>);
......
......@@ -17,11 +17,12 @@
#include "cppa/atom.hpp"
#include "cppa/announce.hpp"
#include "cppa/any_tuple.hpp"
#include "cppa/serializer.hpp"
#include "cppa/deserializer.hpp"
#include "cppa/uniform_type_info.hpp"
#include "cppa/detail/types_array.hpp"
#include "cppa/network/addressed_message.hpp"
#include "cppa/network/message_header.hpp"
#include "cppa/util/callable_trait.hpp"
......@@ -79,7 +80,7 @@ int main() {
// default announced cppa types
"@atom", // cppa::atom_value
"@<>", // cppa::any_tuple
"@msg", // cppa::detail::addressed_message
"@hdr", // cppa::detail::addressed_message
"@actor", // cppa::actor_ptr
"@group", // cppa::group_ptr
"@channel", // cppa::channel_ptr
......@@ -119,7 +120,7 @@ int main() {
std::uint8_t, std::uint16_t, std::uint32_t, std::uint64_t,
std::string, std::u16string, std::u32string,
float, double,
atom_value, any_tuple, network::addressed_message,
atom_value, any_tuple, network::message_header,
actor_ptr, group_ptr,
channel_ptr, intrusive_ptr<process_information>
>::arr;
......@@ -140,7 +141,7 @@ int main() {
uniform_typeid<double>(),
uniform_typeid<atom_value>(),
uniform_typeid<any_tuple>(),
uniform_typeid<network::addressed_message>(),
uniform_typeid<network::message_header>(),
uniform_typeid<actor_ptr>(),
uniform_typeid<group_ptr>(),
uniform_typeid<channel_ptr>(),
......
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