Commit 34086701 authored by Dominik Charousset's avatar Dominik Charousset

Move all IO-related classes & funcs to caf::io

parent acedc6fc
...@@ -32,7 +32,6 @@ ...@@ -32,7 +32,6 @@
#include "caf/channel.hpp" #include "caf/channel.hpp"
#include "caf/message.hpp" #include "caf/message.hpp"
#include "caf/node_id.hpp" #include "caf/node_id.hpp"
#include "caf/publish.hpp"
#include "caf/announce.hpp" #include "caf/announce.hpp"
#include "caf/anything.hpp" #include "caf/anything.hpp"
#include "caf/behavior.hpp" #include "caf/behavior.hpp"
......
...@@ -31,9 +31,7 @@ ...@@ -31,9 +31,7 @@
#include "caf/unit.hpp" #include "caf/unit.hpp"
#include "caf/node_id.hpp" #include "caf/node_id.hpp"
#include "caf/duration.hpp" #include "caf/duration.hpp"
#include "caf/accept_handle.hpp"
#include "caf/system_messages.hpp" #include "caf/system_messages.hpp"
#include "caf/connection_handle.hpp"
#include "caf/detail/type_list.hpp" #include "caf/detail/type_list.hpp"
......
...@@ -19,18 +19,19 @@ ...@@ -19,18 +19,19 @@
#ifndef CAF_ACCEPT_HANDLE_HPP #ifndef CAF_ACCEPT_HANDLE_HPP
#define CAF_ACCEPT_HANDLE_HPP #define CAF_ACCEPT_HANDLE_HPP
#include "caf/io_handle.hpp" #include "caf/io/handle.hpp"
namespace caf { namespace caf {
namespace io {
/** /**
* @brief Generic handle type for managing incoming connections. * @brief Generic handle type for managing incoming connections.
*/ */
class accept_handle : public io_handle<accept_handle> { class accept_handle : public handle<accept_handle> {
friend class io_handle<accept_handle>; friend class handle<accept_handle>;
using super = io_handle<accept_handle>; using super = handle<accept_handle>;
public: public:
...@@ -42,6 +43,7 @@ class accept_handle : public io_handle<accept_handle> { ...@@ -42,6 +43,7 @@ class accept_handle : public io_handle<accept_handle> {
}; };
} // namespace ios
} // namespace caf } // namespace caf
#endif // CAF_ACCEPT_HANDLE_HPP #endif // CAF_ACCEPT_HANDLE_HPP
...@@ -20,12 +20,14 @@ ...@@ -20,12 +20,14 @@
#define CAF_IO_ALL_HPP #define CAF_IO_ALL_HPP
#include "broker.hpp" #include "broker.hpp"
#include "publish.hpp"
#include "spawn_io.hpp" #include "spawn_io.hpp"
#include "middleman.hpp" #include "middleman.hpp"
#include "max_msg_size.hpp" #include "max_msg_size.hpp"
#include "publish_impl.hpp" #include "publish_impl.hpp"
#include "remote_actor.hpp" #include "remote_actor.hpp"
#include "receive_policy.hpp" #include "receive_policy.hpp"
#include "system_messages.hpp"
#include "publish_local_groups.hpp" #include "publish_local_groups.hpp"
#endif // CAF_IO_ALL_HPP #endif // CAF_IO_ALL_HPP
...@@ -25,10 +25,11 @@ ...@@ -25,10 +25,11 @@
#include "caf/spawn.hpp" #include "caf/spawn.hpp"
#include "caf/extend.hpp" #include "caf/extend.hpp"
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
#include "caf/accept_handle.hpp"
#include "caf/connection_handle.hpp"
#include "caf/io/network.hpp" #include "caf/io/network.hpp"
#include "caf/io/accept_handle.hpp"
#include "caf/io/system_messages.hpp"
#include "caf/io/connection_handle.hpp"
#include "caf/mixin/functor_based.hpp" #include "caf/mixin/functor_based.hpp"
#include "caf/mixin/behavior_stack_based.hpp" #include "caf/mixin/behavior_stack_based.hpp"
......
...@@ -19,18 +19,19 @@ ...@@ -19,18 +19,19 @@
#ifndef CAF_CONNECTION_HANDLE_HPP #ifndef CAF_CONNECTION_HANDLE_HPP
#define CAF_CONNECTION_HANDLE_HPP #define CAF_CONNECTION_HANDLE_HPP
#include "caf/io_handle.hpp" #include "caf/io/handle.hpp"
namespace caf { namespace caf {
namespace io {
/** /**
* @brief Generic handle type for identifying connections. * @brief Generic handle type for identifying connections.
*/ */
class connection_handle : public io_handle<connection_handle> { class connection_handle : public handle<connection_handle> {
friend class io_handle<connection_handle>; friend class handle<connection_handle>;
using super = io_handle<connection_handle>; using super = handle<connection_handle>;
public: public:
...@@ -44,6 +45,7 @@ class connection_handle : public io_handle<connection_handle> { ...@@ -44,6 +45,7 @@ class connection_handle : public io_handle<connection_handle> {
constexpr connection_handle invalid_connection_handle = connection_handle{}; constexpr connection_handle invalid_connection_handle = connection_handle{};
} // namespace io
} // namespace caf } // namespace caf
#endif // CAF_CONNECTION_HANDLE_HPP #endif // CAF_CONNECTION_HANDLE_HPP
...@@ -30,17 +30,17 @@ namespace caf { ...@@ -30,17 +30,17 @@ namespace caf {
* {@link connection_handle}. * {@link connection_handle}.
*/ */
template<typename Subtype, int64_t InvalidId = -1> template<typename Subtype, int64_t InvalidId = -1>
class io_handle : detail::comparable<Subtype> { class handle : detail::comparable<Subtype> {
public: public:
constexpr io_handle() : m_id{InvalidId} {} constexpr handle() : m_id{InvalidId} {}
io_handle(const Subtype& other) { m_id = other.id(); } handle(const Subtype& other) { m_id = other.id(); }
io_handle(const io_handle& other) = default; handle(const handle& other) = default;
Subtype& operator=(const io_handle& other) { Subtype& operator=(const handle& other) {
m_id = other.id(); m_id = other.id();
return *static_cast<Subtype*>(this); return *static_cast<Subtype*>(this);
} }
...@@ -67,7 +67,7 @@ class io_handle : detail::comparable<Subtype> { ...@@ -67,7 +67,7 @@ class io_handle : detail::comparable<Subtype> {
protected: protected:
inline io_handle(int64_t handle_id) : m_id{handle_id} {} inline handle(int64_t handle_id) : m_id{handle_id} {}
private: private:
......
...@@ -28,13 +28,13 @@ ...@@ -28,13 +28,13 @@
#include "caf/extend.hpp" #include "caf/extend.hpp"
#include "caf/exception.hpp" #include "caf/exception.hpp"
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
#include "caf/accept_handle.hpp"
#include "caf/connection_handle.hpp"
#include "caf/mixin/memory_cached.hpp" #include "caf/mixin/memory_cached.hpp"
#include "caf/io/fwd.hpp" #include "caf/io/fwd.hpp"
#include "caf/io/accept_handle.hpp"
#include "caf/io/receive_policy.hpp" #include "caf/io/receive_policy.hpp"
#include "caf/io/connection_handle.hpp"
#include "caf/detail/logging.hpp" #include "caf/detail/logging.hpp"
......
...@@ -16,8 +16,8 @@ ...@@ -16,8 +16,8 @@
* accompanying file LICENSE or copy at http://www.boost.org/LICENSE_1_0.txt * * accompanying file LICENSE or copy at http://www.boost.org/LICENSE_1_0.txt *
\******************************************************************************/ \******************************************************************************/
#ifndef CAF_PUBLISH_HPP #ifndef CAF_IO_PUBLISH_HPP
#define CAF_PUBLISH_HPP #define CAF_IO_PUBLISH_HPP
#include <cstdint> #include <cstdint>
...@@ -27,6 +27,7 @@ ...@@ -27,6 +27,7 @@
#include "caf/io/publish_impl.hpp" #include "caf/io/publish_impl.hpp"
namespace caf { namespace caf {
namespace io {
/** /**
* @brief Publishes @p whom at @p port. * @brief Publishes @p whom at @p port.
...@@ -54,6 +55,7 @@ void typed_publish(typed_actor<Rs...> whom, uint16_t port, ...@@ -54,6 +55,7 @@ void typed_publish(typed_actor<Rs...> whom, uint16_t port,
io::publish_impl(actor_cast<abstract_actor_ptr>(whom), port, addr); io::publish_impl(actor_cast<abstract_actor_ptr>(whom), port, addr);
} }
} // namespace io
} // namespace caf } // namespace caf
#endif // CAF_PUBLISH_HPP #endif // CAF_IO_PUBLISH_HPP
...@@ -20,9 +20,10 @@ ...@@ -20,9 +20,10 @@
#define CAF_IO_SPAWN_IO_HPP #define CAF_IO_SPAWN_IO_HPP
#include "caf/spawn.hpp" #include "caf/spawn.hpp"
#include "caf/connection_handle.hpp"
#include "caf/io/broker.hpp" #include "caf/io/broker.hpp"
#include "caf/io/middleman.hpp"
#include "caf/io/connection_handle.hpp"
namespace caf { namespace caf {
namespace io { namespace io {
......
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the Boost Software License, Version 1.0. See *
* accompanying file LICENSE or copy at http://www.boost.org/LICENSE_1_0.txt *
\******************************************************************************/
#ifndef CAF_IO_SYSTEM_MESSAGES_HPP
#define CAF_IO_SYSTEM_MESSAGES_HPP
#include "caf/io/handle.hpp"
#include "caf/io/accept_handle.hpp"
#include "caf/io/connection_handle.hpp"
namespace caf {
namespace io {
/**
* @brief Signalizes a newly accepted connection from a {@link broker}.
*/
struct new_connection_msg {
/**
* @brief The handle that accepted the new connection.
*/
accept_handle source;
/**
* @brief The handle for the new connection.
*/
connection_handle handle;
};
inline bool operator==(const new_connection_msg& lhs,
const new_connection_msg& rhs) {
return lhs.source == rhs.source && lhs.handle == rhs.handle;
}
inline bool operator!=(const new_connection_msg& lhs,
const new_connection_msg& rhs) {
return !(lhs == rhs);
}
/**
* @brief Signalizes newly arrived data for a {@link broker}.
*/
struct new_data_msg {
/**
* @brief Handle to the related connection.
*/
connection_handle handle;
/**
* @brief Buffer containing the received data.
*/
std::vector<char> buf;
};
inline bool operator==(const new_data_msg& lhs, const new_data_msg& rhs) {
return lhs.handle == rhs.handle && lhs.buf == rhs.buf;
}
inline bool operator!=(const new_data_msg& lhs, const new_data_msg& rhs) {
return !(lhs == rhs);
}
/**
* @brief Signalizes that a {@link broker} connection has been closed.
*/
struct connection_closed_msg {
/**
* @brief Handle to the closed connection.
*/
connection_handle handle;
};
inline bool operator==(const connection_closed_msg& lhs,
const connection_closed_msg& rhs) {
return lhs.handle == rhs.handle;
}
inline bool operator!=(const connection_closed_msg& lhs,
const connection_closed_msg& rhs) {
return !(lhs == rhs);
}
/**
* @brief Signalizes that a {@link broker} acceptor has been closed.
*/
struct acceptor_closed_msg {
/**
* @brief Handle to the closed connection.
*/
accept_handle handle;
};
inline bool operator==(const acceptor_closed_msg& lhs,
const acceptor_closed_msg& rhs) {
return lhs.handle == rhs.handle;
}
inline bool operator!=(const acceptor_closed_msg& lhs,
const acceptor_closed_msg& rhs) {
return !(lhs == rhs);
}
} // namespace io
} // namespace caf
#endif // CAF_IO_SYSTEM_MESSAGES_HPP
...@@ -25,8 +25,6 @@ ...@@ -25,8 +25,6 @@
#include "caf/group.hpp" #include "caf/group.hpp"
#include "caf/actor_addr.hpp" #include "caf/actor_addr.hpp"
#include "caf/accept_handle.hpp"
#include "caf/connection_handle.hpp"
#include "caf/detail/tbind.hpp" #include "caf/detail/tbind.hpp"
#include "caf/detail/type_list.hpp" #include "caf/detail/type_list.hpp"
...@@ -155,92 +153,6 @@ inline bool operator!=(const timeout_msg& lhs, const timeout_msg& rhs) { ...@@ -155,92 +153,6 @@ inline bool operator!=(const timeout_msg& lhs, const timeout_msg& rhs) {
return !(lhs == rhs); return !(lhs == rhs);
} }
/**
* @brief Signalizes a newly accepted connection from a {@link broker}.
*/
struct new_connection_msg {
/**
* @brief The handle that accepted the new connection.
*/
accept_handle source;
/**
* @brief The handle for the new connection.
*/
connection_handle handle;
};
inline bool operator==(const new_connection_msg& lhs,
const new_connection_msg& rhs) {
return lhs.source == rhs.source && lhs.handle == rhs.handle;
}
inline bool operator!=(const new_connection_msg& lhs,
const new_connection_msg& rhs) {
return !(lhs == rhs);
}
/**
* @brief Signalizes newly arrived data for a {@link broker}.
*/
struct new_data_msg {
/**
* @brief Handle to the related connection.
*/
connection_handle handle;
/**
* @brief Buffer containing the received data.
*/
std::vector<char> buf;
};
inline bool operator==(const new_data_msg& lhs, const new_data_msg& rhs) {
return lhs.handle == rhs.handle && lhs.buf == rhs.buf;
}
inline bool operator!=(const new_data_msg& lhs, const new_data_msg& rhs) {
return !(lhs == rhs);
}
/**
* @brief Signalizes that a {@link broker} connection has been closed.
*/
struct connection_closed_msg {
/**
* @brief Handle to the closed connection.
*/
connection_handle handle;
};
inline bool operator==(const connection_closed_msg& lhs,
const connection_closed_msg& rhs) {
return lhs.handle == rhs.handle;
}
inline bool operator!=(const connection_closed_msg& lhs,
const connection_closed_msg& rhs) {
return !(lhs == rhs);
}
/**
* @brief Signalizes that a {@link broker} acceptor has been closed.
*/
struct acceptor_closed_msg {
/**
* @brief Handle to the closed connection.
*/
accept_handle handle;
};
inline bool operator==(const acceptor_closed_msg& lhs,
const acceptor_closed_msg& rhs) {
return lhs.handle == rhs.handle;
}
inline bool operator!=(const acceptor_closed_msg& lhs,
const acceptor_closed_msg& rhs) {
return !(lhs == rhs);
}
} // namespace caf } // namespace caf
#endif // CAF_SYSTEM_MESSAGES_HPP #endif // CAF_SYSTEM_MESSAGES_HPP
...@@ -28,13 +28,32 @@ ...@@ -28,13 +28,32 @@
#include "cppa/opt.hpp" #include "cppa/opt.hpp"
#include "cppa/cow_tuple.hpp" #include "cppa/cow_tuple.hpp"
#include "cppa/tuple_cast.hpp" #include "cppa/tuple_cast.hpp"
#include "cppa/max_msg_size.hpp"
#include "cppa/remote_actor.hpp"
#include "cppa/publish_local_groups.hpp" #include "cppa/publish_local_groups.hpp"
// headers for classes and functions that have been moved
#include "caf/io/all.hpp"
// set old namespace to new namespace // set old namespace to new namespace
namespace cppa = caf; namespace cppa = caf;
// re-import a couple of moved things backinto the global CAF namespace
namespace caf {
using io::accept_handle;
using io::connection_handle;
using io::publish;
using io::remote_actor;
using io::max_msg_size;
using io::typed_publish;
using io::typed_remote_actor;
using io::publish_local_groups;
using io::new_connection_msg;
using io::new_data_msg;
using io::connection_closed_msg;
using io::acceptor_closed_msg;
} // namespace caf
// </backward_compatibility> // </backward_compatibility>
#endif // CPPA_CPPA_HPP #endif // CPPA_CPPA_HPP
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the Boost Software License, Version 1.0. See *
* accompanying file LICENSE or copy at http://www.boost.org/LICENSE_1_0.txt *
\******************************************************************************/
#ifndef CPPA_MAX_MSG_SIZE_HPP
#define CPPA_MAX_MSG_SIZE_HPP
// <backward_compatibility version="0.9" whole_file="yes">
#include "caf/io/max_msg_size.hpp"
namespace caf {
using io::max_msg_size;
} // namespace caf
// </backward_compatibility>
#endif // CPPA_MAX_MSG_SIZE_HPP
/******************************************************************************\
* ___ __ *
* /\_ \ __/\ \ *
* \//\ \ /\_\ \ \____ ___ _____ _____ __ *
* \ \ \ \/\ \ \ '__`\ /'___\/\ '__`\/\ '__`\ /'__`\ *
* \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_ *
* /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\ *
* \/____/ \/_/\/___/ \/____/ \ \ \/ \ \ \/ \/__/\/_/ *
* \ \_\ \ \_\ *
* \/_/ \/_/ *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the Boost Software License, Version 1.0. See *
* accompanying file LICENSE or copy at http://www.boost.org/LICENSE_1_0.txt *
\******************************************************************************/
#ifndef CPPA_REMOTE_ACTOR_HPP
#define CPPA_REMOTE_ACTOR_HPP
// <backward_compatibility version="0.9" whole_file="yes">
#include "caf/io/remote_actor.hpp"
namespace caf {
using io::remote_actor;
using io::typed_remote_actor;
} // namespace caf
// </backward_compatibility>
#endif // CPPA_REMOTE_ACTOR_HPP
...@@ -216,7 +216,7 @@ int main(int argc, char** argv) { ...@@ -216,7 +216,7 @@ int main(int argc, char** argv) {
if (mode == "server") { if (mode == "server") {
try { try {
// try to publish math actor at given port // try to publish math actor at given port
publish(spawn(calculator), port); io::publish(spawn(calculator), port);
} }
catch (exception& e) { catch (exception& e) {
cerr << "*** unable to publish math actor at port " << port << "\n" cerr << "*** unable to publish math actor at port " << port << "\n"
......
...@@ -18,7 +18,6 @@ ...@@ -18,7 +18,6 @@
#include "caf/all.hpp" #include "caf/all.hpp"
#include "caf/message.hpp" #include "caf/message.hpp"
#include "caf/publish.hpp"
#include "caf/abstract_group.hpp" #include "caf/abstract_group.hpp"
#include "caf/detail/shared_spinlock.hpp" #include "caf/detail/shared_spinlock.hpp"
......
...@@ -28,11 +28,13 @@ ...@@ -28,11 +28,13 @@
#include "caf/actor.hpp" #include "caf/actor.hpp"
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/node_id.hpp" #include "caf/node_id.hpp"
#include "caf/announce.hpp"
#include "caf/to_string.hpp" #include "caf/to_string.hpp"
#include "caf/actor_proxy.hpp" #include "caf/actor_proxy.hpp"
#include "caf/uniform_type_info.hpp" #include "caf/uniform_type_info.hpp"
#include "caf/io/middleman.hpp" #include "caf/io/middleman.hpp"
#include "caf/io/system_messages.hpp"
#include "caf/io/remote_actor_proxy.hpp" #include "caf/io/remote_actor_proxy.hpp"
#include "caf/detail/logging.hpp" #include "caf/detail/logging.hpp"
...@@ -52,6 +54,142 @@ ...@@ -52,6 +54,142 @@
namespace caf { namespace caf {
namespace io { namespace io {
namespace {
template<typename Subtype>
inline void serialize_impl(const handle<Subtype>& hdl, serializer* sink) {
sink->write_value(hdl.id());
}
template<typename Subtype>
inline void deserialize_impl(handle<Subtype>& hdl, deserializer* source) {
hdl.set_id(source->read<int64_t>());
}
inline void serialize_impl(const new_connection_msg& msg, serializer* sink) {
serialize_impl(msg.source, sink);
serialize_impl(msg.handle, sink);
}
inline void deserialize_impl(new_connection_msg& msg, deserializer* source) {
deserialize_impl(msg.source, source);
deserialize_impl(msg.handle, source);
}
inline void serialize_impl(const new_data_msg& msg, serializer* sink) {
serialize_impl(msg.handle, sink);
auto buf_size = static_cast<uint32_t>(msg.buf.size());
if (buf_size != msg.buf.size()) { // narrowing error
std::ostringstream oss;
oss << "attempted to send more than "
<< std::numeric_limits<uint32_t>::max() << " bytes";
auto errstr = oss.str();
CAF_LOGF_ERROR(errstr);
throw std::ios_base::failure(std::move(errstr));
}
sink->write_value(buf_size);
sink->write_raw(msg.buf.size(), msg.buf.data());
}
inline void deserialize_impl(new_data_msg& msg, deserializer* source) {
deserialize_impl(msg.handle, source);
auto buf_size = source->read<uint32_t>();
msg.buf.resize(buf_size);
source->read_raw(msg.buf.size(), msg.buf.data());
}
// connection_closed_msg & acceptor_closed_msg have the same fields
template<typename T>
typename std::enable_if<std::is_same<T, connection_closed_msg>::value ||
std::is_same<T, acceptor_closed_msg>::value>::type
serialize_impl(const T& dm, serializer* sink) {
serialize_impl(dm.handle, sink);
}
// connection_closed_msg & acceptor_closed_msg have the same fields
template<typename T>
typename std::enable_if<std::is_same<T, connection_closed_msg>::value ||
std::is_same<T, acceptor_closed_msg>::value>::type
deserialize_impl(T& dm, deserializer* source) {
deserialize_impl(dm.handle, source);
}
template<typename T>
class uti_impl : public uniform_type_info {
public:
uti_impl() : m_native(&typeid(T)), m_name(detail::demangle<T>()) {
// nop
}
bool equal_to(const std::type_info& ti) const override {
// in some cases (when dealing with dynamic libraries),
// address can be different although types are equal
return m_native == &ti || *m_native == ti;
}
bool equals(const void* lhs, const void* rhs) const override {
return deref(lhs) == deref(rhs);
}
uniform_value create(const uniform_value& other) const override {
return this->create_impl<T>(other);
}
message as_message(void* instance) const override {
return make_message(deref(instance));
}
const char* name() const {
return m_name.c_str();
}
protected:
void serialize(const void* instance, serializer* sink) const {
serialize_impl(deref(instance), sink);
}
void deserialize(void* instance, deserializer* source) const {
deserialize_impl(deref(instance), source);
}
private:
static inline T& deref(void* ptr) {
return *reinterpret_cast<T*>(ptr);
}
static inline const T& deref(const void* ptr) {
return *reinterpret_cast<const T*>(ptr);
}
const std::type_info* m_native;
std::string m_name;
};
template<typename... Ts>
struct announce_helper;
template<typename T, typename... Ts>
struct announce_helper<T, Ts...> {
static inline void exec() {
announce(typeid(T), uniform_type_info_ptr{new uti_impl<T>});
announce_helper<Ts...>::exec();
}
};
template<>
struct announce_helper<> {
static inline void exec() {
// end of recursion
}
};
} // namespace <anonymous>
using detail::make_counted; using detail::make_counted;
middleman* middleman::instance() { middleman* middleman::instance() {
...@@ -76,6 +214,12 @@ void middleman::initialize() { ...@@ -76,6 +214,12 @@ void middleman::initialize() {
m_backend.run(); m_backend.run();
}); });
m_backend.m_tid = m_thread.get_id(); m_backend.m_tid = m_thread.get_id();
// announce io-related types
announce_helper<new_data_msg, new_connection_msg,
acceptor_closed_msg, connection_closed_msg,
accept_handle, acceptor_closed_msg,
connection_closed_msg, connection_handle,
new_connection_msg, new_data_msg>::exec();
} }
void middleman::stop() { void middleman::stop() {
......
...@@ -18,6 +18,7 @@ ...@@ -18,6 +18,7 @@
#include "caf/all.hpp" #include "caf/all.hpp"
#include "caf/io/publish.hpp"
#include "caf/io/publish_local_groups.hpp" #include "caf/io/publish_local_groups.hpp"
namespace caf { namespace caf {
......
...@@ -53,14 +53,10 @@ namespace detail { ...@@ -53,14 +53,10 @@ namespace detail {
namespace { namespace {
const char* mapped_type_names[][2] = { const char* mapped_type_names[][2] = {
{"bool", "bool" }, {"bool", "bool" },
{"caf::accept_handle", "@accept" },
{"caf::acceptor_closed_msg", "@acceptor_closed" },
{"caf::actor", "@actor" }, {"caf::actor", "@actor" },
{"caf::actor_addr", "@addr" }, {"caf::actor_addr", "@addr" },
{"caf::atom_value", "@atom" }, {"caf::atom_value", "@atom" },
{"caf::channel", "@channel" }, {"caf::channel", "@channel" },
{"caf::connection_closed_msg", "@connection_closed" },
{"caf::connection_handle", "@connection" },
{"caf::down_msg", "@down" }, {"caf::down_msg", "@down" },
{"caf::duration", "@duration" }, {"caf::duration", "@duration" },
{"caf::exit_msg", "@exit" }, {"caf::exit_msg", "@exit" },
...@@ -68,16 +64,14 @@ const char* mapped_type_names[][2] = { ...@@ -68,16 +64,14 @@ const char* mapped_type_names[][2] = {
{"caf::group_down_msg", "@group_down" }, {"caf::group_down_msg", "@group_down" },
{"caf::message", "@message" }, {"caf::message", "@message" },
{"caf::message_id", "@message_id" }, {"caf::message_id", "@message_id" },
{"caf::new_connection_msg", "@new_connection" },
{"caf::new_data_msg", "@new_data" },
{"caf::node_id", "@node" }, {"caf::node_id", "@node" },
{"caf::sync_exited_msg", "@sync_exited" }, {"caf::sync_exited_msg", "@sync_exited" },
{"caf::sync_timeout_msg", "@sync_timeout" }, {"caf::sync_timeout_msg", "@sync_timeout" },
{"caf::timeout_msg", "@timeout" }, {"caf::timeout_msg", "@timeout" },
{"caf::unit_t", "@unit" }, {"caf::unit_t", "@unit" },
{"double", "double" }, {"double", "double" },
{"float", "float" }, {"float", "float" },
{"long double", "@ldouble" }, {"long double", "@ldouble" },
// std::string // std::string
{"std::basic_string<@i8,std::char_traits<@i8>,std::allocator<@i8>>", {"std::basic_string<@i8,std::char_traits<@i8>,std::allocator<@i8>>",
"@str"}, "@str"},
...@@ -96,14 +90,10 @@ const char* mapped_type_names[][2] = { ...@@ -96,14 +90,10 @@ const char* mapped_type_names[][2] = {
}; };
// the order of this table must be *identical* to mapped_type_names // the order of this table must be *identical* to mapped_type_names
using static_type_table = type_list<bool, using static_type_table = type_list<bool,
accept_handle,
acceptor_closed_msg,
actor, actor,
actor_addr, actor_addr,
atom_value, atom_value,
channel, channel,
connection_closed_msg,
connection_handle,
down_msg, down_msg,
duration, duration,
exit_msg, exit_msg,
...@@ -111,8 +101,6 @@ using static_type_table = type_list<bool, ...@@ -111,8 +101,6 @@ using static_type_table = type_list<bool,
group_down_msg, group_down_msg,
message, message,
message_id, message_id,
new_connection_msg,
new_data_msg,
node_id, node_id,
sync_exited_msg, sync_exited_msg,
sync_timeout_msg, sync_timeout_msg,
...@@ -447,64 +435,6 @@ inline void serialize_impl(const sync_timeout_msg&, serializer*) {} ...@@ -447,64 +435,6 @@ inline void serialize_impl(const sync_timeout_msg&, serializer*) {}
inline void deserialize_impl(const sync_timeout_msg&, deserializer*) {} inline void deserialize_impl(const sync_timeout_msg&, deserializer*) {}
template<typename Subtype>
inline void serialize_impl(const io_handle<Subtype>& hdl, serializer* sink) {
sink->write_value(hdl.id());
}
template<typename Subtype>
inline void deserialize_impl(io_handle<Subtype>& hdl, deserializer* source) {
hdl.set_id(source->read<int64_t>());
}
inline void serialize_impl(const new_connection_msg& msg, serializer* sink) {
serialize_impl(msg.source, sink);
serialize_impl(msg.handle, sink);
}
inline void deserialize_impl(new_connection_msg& msg, deserializer* source) {
deserialize_impl(msg.source, source);
deserialize_impl(msg.handle, source);
}
inline void serialize_impl(const new_data_msg& msg, serializer* sink) {
serialize_impl(msg.handle, sink);
auto buf_size = static_cast<uint32_t>(msg.buf.size());
if (buf_size != msg.buf.size()) { // narrowing error
std::ostringstream oss;
oss << "attempted to send more than "
<< std::numeric_limits<uint32_t>::max() << " bytes";
auto errstr = oss.str();
CAF_LOGF_ERROR(errstr);
throw std::ios_base::failure(std::move(errstr));
}
sink->write_value(buf_size);
sink->write_raw(msg.buf.size(), msg.buf.data());
}
inline void deserialize_impl(new_data_msg& msg, deserializer* source) {
deserialize_impl(msg.handle, source);
auto buf_size = source->read<uint32_t>();
msg.buf.resize(buf_size);
source->read_raw(msg.buf.size(), msg.buf.data());
}
// exit_msg & down_msg have the same fields
template<typename T>
typename std::enable_if<std::is_same<T, connection_closed_msg>::value ||
std::is_same<T, acceptor_closed_msg>::value>::type
serialize_impl(const T& dm, serializer* sink) {
serialize_impl(dm.handle, sink);
}
// exit_msg & down_msg have the same fields
template<typename T>
typename std::enable_if<std::is_same<T, connection_closed_msg>::value ||
std::is_same<T, acceptor_closed_msg>::value>::type
deserialize_impl(T& dm, deserializer* source) {
deserialize_impl(dm.handle, source);
}
bool types_equal(const std::type_info* lhs, const std::type_info* rhs) { bool types_equal(const std::type_info* lhs, const std::type_info* rhs) {
// in some cases (when dealing with dynamic libraries), // in some cases (when dealing with dynamic libraries),
// address can be different although types are equal // address can be different although types are equal
...@@ -899,13 +829,7 @@ class utim_impl : public uniform_type_info_map { ...@@ -899,13 +829,7 @@ class utim_impl : public uniform_type_info_map {
int_tinfo<uint32_t>, int_tinfo<uint32_t>,
int_tinfo<int64_t>, int_tinfo<int64_t>,
int_tinfo<uint64_t>, int_tinfo<uint64_t>,
default_uniform_type_info<charbuf>, default_uniform_type_info<charbuf>>;
uti_impl<accept_handle>,
uti_impl<connection_handle>,
uti_impl<acceptor_closed_msg>,
uti_impl<connection_closed_msg>,
uti_impl<new_connection_msg>,
uti_impl<new_data_msg>>;
builtin_types m_storage; builtin_types m_storage;
......
...@@ -318,7 +318,7 @@ void test_remote_actor(std::string app_path, bool run_remote_actor) { ...@@ -318,7 +318,7 @@ void test_remote_actor(std::string app_path, bool run_remote_actor) {
bool success = false; bool success = false;
do { do {
try { try {
publish(serv, port, "127.0.0.1"); io::publish(serv, port, "127.0.0.1");
success = true; success = true;
CAF_PRINT("running on port " << port); CAF_PRINT("running on port " << port);
CAF_LOGF_INFO("running on port " << port); CAF_LOGF_INFO("running on port " << port);
......
...@@ -74,7 +74,7 @@ uint16_t run_server() { ...@@ -74,7 +74,7 @@ uint16_t run_server() {
uint16_t port = 4242; uint16_t port = 4242;
for (;;) { for (;;) {
try { try {
typed_publish(ref, port, "127.0.0.1"); io::typed_publish(ref, port, "127.0.0.1");
CAF_LOGF_DEBUG("running on port " << port); CAF_LOGF_DEBUG("running on port " << port);
return port; return port;
} }
......
...@@ -43,6 +43,65 @@ inline bool operator==(const foo& lhs, const foo& rhs) { ...@@ -43,6 +43,65 @@ inline bool operator==(const foo& lhs, const foo& rhs) {
using namespace caf; using namespace caf;
bool check_types(const std::set<std::string>& expected) {
// holds the type names we see at runtime
std::set<std::string> found;
// fetch all available type names
auto types = uniform_type_info::instances();
for (auto tinfo : types) {
found.insert(tinfo->name());
}
// compare the two sets
CAF_CHECK_EQUAL(expected.size(), found.size());
bool expected_equals_found = false;
if (expected.size() == found.size()
&& std::equal(found.begin(), found.end(), expected.begin())) {
CAF_CHECKPOINT();
return true;
}
CAF_CHECK(false);
if (!expected_equals_found) {
std::string(41, ' ');
std::ostringstream oss(std::string(41, ' '));
oss.seekp(0);
oss << "found (" << found.size() << ")";
oss.seekp(22);
oss << "expected (" << expected.size() << ")";
std::string lhs;
std::string rhs;
CAF_PRINT(oss.str());
CAF_PRINT(std::string(41, '-'));
auto fi = found.begin();
auto fe = found.end();
auto ei = expected.begin();
auto ee = expected.end();
while (fi != fe || ei != ee) {
if (fi != fe)
lhs = *fi++;
else
lhs.clear();
if (ei != ee)
rhs = *ei++;
else
rhs.clear();
lhs.resize(20, ' ');
CAF_PRINT(lhs << "| " << rhs);
}
}
return false;
}
template<typename T>
T& append(T& storage) {
return storage;
}
template<typename T, typename U, typename... Us>
T& append(T& storage, U&& u, Us&&... us) {
storage.insert(std::forward<U>(u));
return append(storage, std::forward<Us>(us)...);
}
int main() { int main() {
CAF_TEST(test_uniform_type); CAF_TEST(test_uniform_type);
auto announce1 = announce<foo>(&foo::value); auto announce1 = announce<foo>(&foo::value);
...@@ -84,15 +143,11 @@ int main() { ...@@ -84,15 +143,11 @@ int main() {
"float", "double", "@ldouble", // floating points "float", "double", "@ldouble", // floating points
// default announced types // default announced types
"@unit", // unit_t "@unit", // unit_t
"@accept", // accept_handle
"@acceptor_closed", // acceptor_closed_msg
"@actor", // actor "@actor", // actor
"@addr", // actor_addr "@addr", // actor_addr
"@atom", // atom_value "@atom", // atom_value
"@channel", // channel "@channel", // channel
"@charbuf", // vector<char> "@charbuf", // vector<char>
"@connection", // connection_handle
"@connection_closed", // connection_closed_msg
"@down", // down_msg "@down", // down_msg
"@duration", // duration "@duration", // duration
"@exit", // exit_msg "@exit", // exit_msg
...@@ -100,56 +155,23 @@ int main() { ...@@ -100,56 +155,23 @@ int main() {
"@group_down", // group_down_msg "@group_down", // group_down_msg
"@message", // message "@message", // message
"@message_id", // message_id "@message_id", // message_id
"@new_connection", // new_connection_msg
"@new_data", // new_data_msg
"@node", // node_id "@node", // node_id
"@strmap", // map<string,string> "@strmap", // map<string,string>
"@timeout", // timeout_msg
"@sync_exited", // sync_exited_msg "@sync_exited", // sync_exited_msg
"@sync_timeout", // sync_timeout_msg "@sync_timeout" // sync_timeout_msg
"@timeout" // timeout_msg
}; };
// holds the type names we see at runtime if (check_types(expected)) {
std::set<std::string> found; // causes the middleman to create its singleton
// fetch all available type names io::middleman::instance();
auto types = uniform_type_info::instances(); // ok, check whether middleman announces its types correctly
for (auto tinfo : types) { check_types(append(expected,
found.insert(tinfo->name()); "caf::io::accept_handle",
} "caf::io::acceptor_closed_msg",
// compare the two sets "caf::io::connection_handle",
CAF_CHECK_EQUAL(expected.size(), found.size()); "caf::io::connection_closed_msg",
bool expected_equals_found = false; "caf::io::new_connection_msg",
if (expected.size() == found.size()) { "caf::io::new_data_msg"));
expected_equals_found =
std::equal(found.begin(), found.end(), expected.begin());
CAF_CHECK(expected_equals_found);
}
if (!expected_equals_found) {
std::string(41, ' ');
std::ostringstream oss(std::string(41, ' '));
oss.seekp(0);
oss << "found (" << found.size() << ")";
oss.seekp(22);
oss << "expected (" << found.size() << ")";
std::string lhs;
std::string rhs;
CAF_PRINT(oss.str());
CAF_PRINT(std::string(41, '-'));
auto fi = found.begin();
auto fe = found.end();
auto ei = expected.begin();
auto ee = expected.end();
while (fi != fe || ei != ee) {
if (fi != fe)
lhs = *fi++;
else
lhs.clear();
if (ei != ee)
rhs = *ei++;
else
rhs.clear();
lhs.resize(20, ' ');
CAF_PRINT(lhs << "| " << rhs);
}
} }
return CAF_TEST_RESULT(); return CAF_TEST_RESULT();
} }
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