Commit 458f9b29 authored by Dominik Charousset's avatar Dominik Charousset

renamed namespace `network` to `io`

parent a15df24a
......@@ -107,11 +107,13 @@ set(LIBCPPA_SRC
src/behavior_stack.cpp
src/binary_deserializer.cpp
src/binary_serializer.cpp
src/broker.cpp
src/broker_backend.cpp
src/buffer.cpp
src/buffered_writer.cpp
src/channel.cpp
src/context_switching_actor.cpp
src/continuable_io.cpp
src/continuable.cpp
src/decorated_tuple.cpp
src/default_actor_addressing.cpp
src/default_actor_proxy.cpp
......@@ -132,8 +134,6 @@ set(LIBCPPA_SRC
src/get_mac_addresses.cpp
src/group.cpp
src/group_manager.cpp
src/io_actor.cpp
src/io_actor_backend.cpp
src/io_service.cpp
src/ipv4_acceptor.cpp
src/ipv4_io_stream.cpp
......
......@@ -79,6 +79,26 @@ cppa/guard_expr.hpp
cppa/intrusive/blocking_single_reader_queue.hpp
cppa/intrusive/single_reader_queue.hpp
cppa/intrusive_ptr.hpp
cppa/io/acceptor.hpp
cppa/io/broker.hpp
cppa/io/broker_backend.hpp
cppa/io/buffered_writer.hpp
cppa/io/continuable.hpp
cppa/io/default_actor_addressing.hpp
cppa/io/default_actor_proxy.hpp
cppa/io/default_message_queue.hpp
cppa/io/default_peer.hpp
cppa/io/default_peer_acceptor.hpp
cppa/io/default_protocol.hpp
cppa/io/input_stream.hpp
cppa/io/io_handle.hpp
cppa/io/ipv4_acceptor.hpp
cppa/io/ipv4_io_stream.hpp
cppa/io/middleman.hpp
cppa/io/middleman_event_handler.hpp
cppa/io/output_stream.hpp
cppa/io/protocol.hpp
cppa/io/stream.hpp
cppa/local_actor.hpp
cppa/logging.hpp
cppa/mailbox_based.hpp
......@@ -91,26 +111,6 @@ cppa/message_future.hpp
cppa/message_header.hpp
cppa/message_id.hpp
cppa/message_priority.hpp
cppa/network/acceptor.hpp
cppa/network/buffered_writer.hpp
cppa/network/continuable_io.hpp
cppa/network/default_actor_addressing.hpp
cppa/network/default_actor_proxy.hpp
cppa/network/default_message_queue.hpp
cppa/network/default_peer.hpp
cppa/network/default_peer_acceptor.hpp
cppa/network/default_protocol.hpp
cppa/network/input_stream.hpp
cppa/network/io_actor.hpp
cppa/network/io_actor_backend.hpp
cppa/network/io_service.hpp
cppa/network/io_stream.hpp
cppa/network/ipv4_acceptor.hpp
cppa/network/ipv4_io_stream.hpp
cppa/network/middleman.hpp
cppa/network/middleman_event_handler.hpp
cppa/network/output_stream.hpp
cppa/network/protocol.hpp
cppa/object.hpp
cppa/on.hpp
cppa/opencl.hpp
......@@ -217,11 +217,13 @@ src/behavior.cpp
src/behavior_stack.cpp
src/binary_deserializer.cpp
src/binary_serializer.cpp
src/broker.cpp
src/broker_backend.cpp
src/buffer.cpp
src/buffered_writer.cpp
src/channel.cpp
src/context_switching_actor.cpp
src/continuable_io.cpp
src/continuable.cpp
src/decorated_tuple.cpp
src/default_actor_addressing.cpp
src/default_actor_proxy.cpp
......@@ -242,8 +244,6 @@ src/get_mac_addresses.cpp
src/get_root_uuid.cpp
src/group.cpp
src/group_manager.cpp
src/io_actor.cpp
src/io_actor_backend.cpp
src/io_service.cpp
src/ipv4_acceptor.cpp
src/ipv4_io_stream.cpp
......
......@@ -67,13 +67,13 @@
#include "cppa/event_based_actor.hpp"
#include "cppa/util/type_traits.hpp"
#include "cppa/network/acceptor.hpp"
#include "cppa/network/io_actor.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/network/io_service.hpp"
#include "cppa/network/input_stream.hpp"
#include "cppa/network/output_stream.hpp"
#include "cppa/network/io_actor_backend.hpp"
#include "cppa/io/acceptor.hpp"
#include "cppa/io/broker.hpp"
#include "cppa/io/middleman.hpp"
#include "cppa/io/io_handle.hpp"
#include "cppa/io/input_stream.hpp"
#include "cppa/io/output_stream.hpp"
#include "cppa/io/broker_backend.hpp"
#include "cppa/detail/memory.hpp"
#include "cppa/detail/get_behavior.hpp"
......@@ -585,7 +585,7 @@ void publish(actor_ptr whom, std::uint16_t port, const char* addr = nullptr);
* @param whom Actor that should be published at @p port.
* @param acceptor Network technology-specific acceptor implementation.
*/
void publish(actor_ptr whom, std::unique_ptr<network::acceptor> acceptor);
void publish(actor_ptr whom, std::unique_ptr<io::acceptor> acceptor);
/**
* @brief Establish a new connection to the actor at @p host on given @p port.
......@@ -610,23 +610,23 @@ inline actor_ptr remote_actor(const std::string& host, std::uint16_t port) {
* @returns An {@link actor_ptr} to the proxy instance
* representing a remote actor.
*/
actor_ptr remote_actor(network::io_stream_ptr_pair connection);
actor_ptr remote_actor(io::stream_ptr_pair connection);
/**
* @brief Spawns an IO actor of type @p Impl.
* @param args Constructor arguments.
* @tparam Impl Subtype of {@link network::io_actor}.
* @tparam Impl Subtype of {@link io::broker}.
* @tparam Options Optional flags to modify <tt>spawn</tt>'s behavior.
* @returns An {@link actor_ptr} to the spawned {@link actor}.
*/
template<class Impl, spawn_options Options = no_spawn_options, typename... Ts>
actor_ptr spawn_io(network::input_stream_ptr in,
network::output_stream_ptr out,
actor_ptr spawn_io(io::input_stream_ptr in,
io::output_stream_ptr out,
Ts&&... args) {
using namespace network;
using namespace io;
auto mm = get_middleman();
auto ptr = make_counted<Impl>(std::forward<Ts>(args)...);
auto backend = make_counted<io_actor_backend>(std::move(in), std::move(out), ptr);
auto backend = make_counted<broker_backend>(std::move(in), std::move(out), ptr);
backend->init();
mm->run_later([=] { mm->continue_reader(backend); });
return eval_sopts(Options, std::move(ptr));
......@@ -639,16 +639,16 @@ actor_ptr spawn_io(network::input_stream_ptr in,
* @returns An {@link actor_ptr} to the spawned {@link actor}.
*/
template<spawn_options Options = no_spawn_options,
typename F = std::function<void (network::io_service*)>,
typename F = std::function<void (io::io_handle*)>,
typename... Ts>
actor_ptr spawn_io(F fun,
network::input_stream_ptr in,
network::output_stream_ptr out,
io::input_stream_ptr in,
io::output_stream_ptr out,
Ts&&... args) {
using namespace network;
using namespace io;
auto mm = get_middleman();
auto ptr = io_actor::from(std::move(fun), std::forward<Ts>(args)...);
auto backend = make_counted<io_actor_backend>(std::move(in), std::move(out), ptr);
auto ptr = broker::from(std::move(fun), std::forward<Ts>(args)...);
auto backend = make_counted<broker_backend>(std::move(in), std::move(out), ptr);
backend->init();
mm->run_later([=] { mm->continue_reader(backend); });
return eval_sopts(Options, std::move(ptr));
......
......@@ -40,7 +40,7 @@ class scheduler;
} // namespace cppa
namespace cppa { namespace network { class middleman; } }
namespace cppa { namespace io { class middleman; } }
namespace cppa { namespace opencl { class opencl_metainfo; } }
......@@ -70,7 +70,7 @@ class singleton_manager {
static actor_registry* get_actor_registry();
static network::middleman* get_middleman();
static io::middleman* get_middleman();
static uniform_type_info_map* get_uniform_type_info_map();
......
......@@ -36,15 +36,15 @@
#include "cppa/config.hpp"
#include "cppa/option.hpp"
#include "cppa/network/input_stream.hpp"
#include "cppa/network/output_stream.hpp"
#include "cppa/io/input_stream.hpp"
#include "cppa/io/output_stream.hpp"
namespace cppa { namespace network {
namespace cppa { namespace io {
/**
* @brief A pair of input and output stream pointers.
*/
typedef std::pair<input_stream_ptr, output_stream_ptr> io_stream_ptr_pair;
typedef std::pair<input_stream_ptr, output_stream_ptr> stream_ptr_pair;
/**
* @brief Accepts connections from client processes.
......@@ -65,13 +65,13 @@ class acceptor {
* @brief Accepts a new connection and returns an input/output stream pair.
* @note This member function blocks until a new connection was established.
*/
virtual io_stream_ptr_pair accept_connection() = 0;
virtual stream_ptr_pair accept_connection() = 0;
/**
* @brief Tries to accept a new connection but immediately returns if
* there is no pending connection.
*/
virtual option<io_stream_ptr_pair> try_accept_connection() = 0;
virtual option<stream_ptr_pair> try_accept_connection() = 0;
};
......
......@@ -38,21 +38,21 @@
#include "cppa/local_actor.hpp"
#include "cppa/mailbox_element.hpp"
#include "cppa/network/io_service.hpp"
#include "cppa/io/io_handle.hpp"
#include "cppa/detail/fwd.hpp"
namespace cppa { namespace network {
namespace cppa { namespace io {
class io_actor_backend;
class io_actor_continuation;
class broker_backend;
class broker_continuation;
class io_actor : public extend<local_actor>::with<threadless, stackless> {
class broker : public extend<local_actor>::with<threadless, stackless> {
typedef combined_type super;
friend class io_actor_backend;
friend class io_actor_continuation;
friend class broker_backend;
friend class broker_continuation;
public:
......@@ -62,10 +62,10 @@ class io_actor : public extend<local_actor>::with<threadless, stackless> {
void quit(std::uint32_t reason);
static intrusive_ptr<io_actor> from(std::function<void (io_service*)> fun);
static intrusive_ptr<broker> from(std::function<void (io_handle*)> fun);
template<typename F, typename T0, typename... Ts>
static intrusive_ptr<io_actor> from(F fun, T0&& arg0, Ts&&... args) {
static intrusive_ptr<broker> from(F fun, T0&& arg0, Ts&&... args) {
return from(std::bind(std::move(fun),
std::placeholders::_1,
detail::fwd<T0>(arg0),
......@@ -74,7 +74,7 @@ class io_actor : public extend<local_actor>::with<threadless, stackless> {
protected:
io_service& io_handle();
io_handle& io();
private:
......@@ -82,11 +82,11 @@ class io_actor : public extend<local_actor>::with<threadless, stackless> {
void invoke_message(any_tuple msg);
intrusive_ptr<io_actor_backend> m_parent;
intrusive_ptr<broker_backend> m_parent;
};
typedef intrusive_ptr<io_actor> io_actor_ptr;
typedef intrusive_ptr<broker> broker_ptr;
} } // namespace cppa::network
......
......@@ -31,15 +31,17 @@
#ifndef IO_ACTOR_BACKEND_HPP
#define IO_ACTOR_BACKEND_HPP
#include "cppa/network/io_actor.hpp"
#include "cppa/network/io_service.hpp"
#include "cppa/network/input_stream.hpp"
#include "cppa/network/output_stream.hpp"
#include "cppa/network/buffered_writer.hpp"
#include <cstdint>
namespace cppa { namespace network {
#include "cppa/io/broker.hpp"
#include "cppa/io/io_handle.hpp"
#include "cppa/io/input_stream.hpp"
#include "cppa/io/output_stream.hpp"
#include "cppa/io/buffered_writer.hpp"
class io_actor_backend : public buffered_writer, public io_service {
namespace cppa { namespace io {
class broker_backend : public buffered_writer, public io_handle {
typedef buffered_writer super; // io_service is merely an interface
......@@ -48,9 +50,7 @@ class io_actor_backend : public buffered_writer, public io_service {
public:
io_actor_backend(input_stream_ptr in, output_stream_ptr out, io_actor_ptr ptr);
~io_actor_backend();
broker_backend(input_stream_ptr in, output_stream_ptr out, broker_ptr ptr);
void init();
......@@ -72,8 +72,8 @@ class io_actor_backend : public buffered_writer, public io_service {
policy_flag m_policy;
size_t m_policy_buffer_size;
input_stream_ptr m_in;
intrusive_ptr<io_actor> m_self;
cow_tuple<atom_value, util::buffer> m_read;
intrusive_ptr<broker> m_self;
cow_tuple<atom_value, uint32_t, util::buffer> m_read;
};
......
......@@ -33,16 +33,16 @@
#include "cppa/util/buffer.hpp"
#include "cppa/network/output_stream.hpp"
#include "cppa/network/continuable_io.hpp"
#include "cppa/io/output_stream.hpp"
#include "cppa/io/continuable.hpp"
namespace cppa { namespace network {
namespace cppa { namespace io {
class middleman;
class buffered_writer : public continuable_io {
class buffered_writer : public continuable {
typedef continuable_io super;
typedef continuable super;
public:
......
......@@ -35,11 +35,11 @@
#include "cppa/ref_counted.hpp"
#include "cppa/intrusive_ptr.hpp"
namespace cppa { namespace network {
namespace cppa { namespace io {
/**
* @brief Denotes the return value of
* {@link continuable_io::continue_reading()}.
* {@link continuable::continue_reading()}.
*/
enum continue_reading_result {
read_failure,
......@@ -49,7 +49,7 @@ enum continue_reading_result {
/**
* @brief Denotes the return value of
* {@link continuable_io::continue_writing()}.
* {@link continuable::continue_writing()}.
*/
enum continue_writing_result {
write_failure,
......@@ -61,7 +61,7 @@ enum continue_writing_result {
/**
* @brief An object performing asynchronous input and output.
*/
class continuable_io : public ref_counted {
class continuable : public ref_counted {
public:
......@@ -95,7 +95,7 @@ class continuable_io : public ref_counted {
protected:
continuable_io(native_socket_type read_fd,
continuable(native_socket_type read_fd,
native_socket_type write_fd = invalid_socket);
private:
......@@ -105,7 +105,7 @@ class continuable_io : public ref_counted {
};
typedef intrusive_ptr<continuable_io> continuable_io_ptr;
typedef intrusive_ptr<continuable> continuable_ptr;
} } // namespace cppa::network
......
......@@ -38,7 +38,7 @@
#include "cppa/actor_addressing.hpp"
#include "cppa/process_information.hpp"
namespace cppa { namespace network {
namespace cppa { namespace io {
class default_protocol;
......
......@@ -34,7 +34,7 @@
#include "cppa/extend.hpp"
#include "cppa/actor_proxy.hpp"
#include "cppa/memory_cached.hpp"
#include "cppa/network/default_protocol.hpp"
#include "cppa/io/default_protocol.hpp"
#include "cppa/intrusive/single_reader_queue.hpp"
namespace cppa { namespace detail {
......@@ -46,7 +46,7 @@ class basic_memory_cache;
} } // namespace cppa::detail
namespace cppa { namespace network {
namespace cppa { namespace io {
class sync_request_info : public extend<memory_managed>::with<memory_cached> {
......
......@@ -35,7 +35,7 @@
#include "cppa/ref_counted.hpp"
#include "cppa/message_header.hpp"
namespace cppa { namespace network {
namespace cppa { namespace io {
class default_message_queue : public ref_counted {
......
......@@ -41,12 +41,12 @@
#include "cppa/util/buffer.hpp"
#include "cppa/network/input_stream.hpp"
#include "cppa/network/output_stream.hpp"
#include "cppa/network/buffered_writer.hpp"
#include "cppa/network/default_message_queue.hpp"
#include "cppa/io/input_stream.hpp"
#include "cppa/io/output_stream.hpp"
#include "cppa/io/buffered_writer.hpp"
#include "cppa/io/default_message_queue.hpp"
namespace cppa { namespace network {
namespace cppa { namespace io {
class default_protocol;
......
......@@ -33,16 +33,16 @@
#include "cppa/actor.hpp"
#include "cppa/network/ipv4_acceptor.hpp"
#include "cppa/network/continuable_io.hpp"
#include "cppa/io/ipv4_acceptor.hpp"
#include "cppa/io/continuable.hpp"
namespace cppa { namespace network {
namespace cppa { namespace io {
class default_protocol;
class default_peer_acceptor : public continuable_io {
class default_peer_acceptor : public continuable {
typedef continuable_io super;
typedef continuable super;
public:
......
......@@ -37,14 +37,14 @@
#include "cppa/actor_addressing.hpp"
#include "cppa/process_information.hpp"
#include "cppa/network/protocol.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/network/default_peer.hpp"
#include "cppa/network/default_peer_acceptor.hpp"
#include "cppa/network/default_message_queue.hpp"
#include "cppa/network/default_actor_addressing.hpp"
#include "cppa/io/protocol.hpp"
#include "cppa/io/middleman.hpp"
#include "cppa/io/default_peer.hpp"
#include "cppa/io/default_peer_acceptor.hpp"
#include "cppa/io/default_message_queue.hpp"
#include "cppa/io/default_actor_addressing.hpp"
namespace cppa { namespace network {
namespace cppa { namespace io {
class default_protocol : public protocol {
......@@ -66,7 +66,7 @@ class default_protocol : public protocol {
actor_ptr remote_actor(variant_args args);
actor_ptr remote_actor(io_stream_ptr_pair ioptrs, variant_args args);
actor_ptr remote_actor(stream_ptr_pair ioptrs, variant_args args);
void register_peer(const process_information& node, default_peer* ptr);
......
......@@ -35,7 +35,7 @@
#include "cppa/ref_counted.hpp"
#include "cppa/intrusive_ptr.hpp"
namespace cppa { namespace network {
namespace cppa { namespace io {
/**
* @brief An abstract input stream interface.
......
......@@ -33,13 +33,13 @@
#include <cstddef>
namespace cppa { namespace network {
namespace cppa { namespace io {
class io_service {
class io_handle {
public:
virtual ~io_service();
virtual ~io_handle();
/**
* @brief Denotes when an actor will receive a read buffer.
......
......@@ -35,9 +35,9 @@
#include <cstdint>
#include "cppa/config.hpp"
#include "cppa/network/acceptor.hpp"
#include "cppa/io/acceptor.hpp"
namespace cppa { namespace network {
namespace cppa { namespace io {
class ipv4_acceptor : public acceptor {
......@@ -50,9 +50,9 @@ class ipv4_acceptor : public acceptor {
native_socket_type file_handle() const;
io_stream_ptr_pair accept_connection();
stream_ptr_pair accept_connection();
option<io_stream_ptr_pair> try_accept_connection();
option<stream_ptr_pair> try_accept_connection();
private:
......
......@@ -32,17 +32,17 @@
#define CPPA_IPV4_IO_STREAM_HPP
#include "cppa/config.hpp"
#include "cppa/network/io_stream.hpp"
#include "cppa/io/stream.hpp"
namespace cppa { namespace network {
namespace cppa { namespace io {
class ipv4_io_stream : public io_stream {
class ipv4_io_stream : public stream {
public:
static io_stream_ptr connect_to(const char* host, std::uint16_t port);
static stream_ptr connect_to(const char* host, std::uint16_t port);
static io_stream_ptr from_native_socket(native_socket_type fd);
static stream_ptr from_native_socket(native_socket_type fd);
native_socket_type read_handle() const;
......
......@@ -36,11 +36,11 @@
#include <memory>
#include <functional>
#include "cppa/network/continuable_io.hpp"
#include "cppa/io/continuable.hpp"
namespace cppa { namespace detail { class singleton_manager; } }
namespace cppa { namespace network {
namespace cppa { namespace io {
class protocol;
class middleman_impl;
......@@ -70,25 +70,25 @@ class middleman {
* @brief Removes @p ptr from the list of active writers.
* @warning This member function is not thread-safe.
*/
void stop_writer(const continuable_io_ptr& ptr);
void stop_writer(const continuable_ptr& ptr);
/**
* @brief Adds @p ptr to the list of active writers.
* @warning This member function is not thread-safe.
*/
void continue_writer(const continuable_io_ptr& ptr);
void continue_writer(const continuable_ptr& ptr);
/**
* @brief Removes @p ptr from the list of active readers.
* @warning This member function is not thread-safe.
*/
void stop_reader(const continuable_io_ptr& ptr);
void stop_reader(const continuable_ptr& ptr);
/**
* @brief Adds @p ptr to the list of active readers.
* @warning This member function is not thread-safe.
*/
void continue_reader(const continuable_io_ptr& ptr);
void continue_reader(const continuable_ptr& ptr);
protected:
......@@ -103,7 +103,7 @@ class middleman {
// sets m_impl and binds implementation to given protocol
void set_pimpl(std::unique_ptr<protocol>&&);
// creates a middleman using network::default_protocol
// creates a middleman using io::default_protocol
static middleman* create_singleton();
// destroys uninitialized instances
......
......@@ -37,9 +37,9 @@
#include "cppa/config.hpp"
#include "cppa/logging.hpp"
#include "cppa/network/continuable_io.hpp"
#include "cppa/io/continuable.hpp"
namespace cppa { namespace network {
namespace cppa { namespace io {
typedef int event_bitmask;
......@@ -65,10 +65,10 @@ inline event_bitmask from_int_bitmask(unsigned mask) {
struct fd_meta_info {
native_socket_type fd;
continuable_io_ptr ptr;
continuable_ptr ptr;
event_bitmask mask;
fd_meta_info(native_socket_type a0,
const continuable_io_ptr& a1,
const continuable_ptr& a1,
event_bitmask a2)
: fd(a0), ptr(a1), mask(a2) { }
};
......@@ -84,12 +84,12 @@ class middleman_event_handler {
/**
* @brief Enqueues an add operation.
*/
void add_later(const continuable_io_ptr& ptr, event_bitmask e);
void add_later(const continuable_ptr& ptr, event_bitmask e);
/**
* @brief Enqueues an erase operation.
*/
void erase_later(const continuable_io_ptr& ptr, event_bitmask e);
void erase_later(const continuable_ptr& ptr, event_bitmask e);
/**
* @brief Poll all events.
......@@ -123,7 +123,7 @@ class middleman_event_handler {
std::vector<std::pair<fd_meta_info, fd_meta_event>> m_alterations;
std::vector<std::pair<event_bitmask, continuable_io*>> m_events;
std::vector<std::pair<event_bitmask, continuable*>> m_events;
middleman_event_handler();
......@@ -134,11 +134,11 @@ class middleman_event_handler {
native_socket_type fd,
event_bitmask old_bitmask,
event_bitmask new_bitmask,
continuable_io* ptr) = 0;
continuable* ptr) = 0;
private:
void alteration(const continuable_io_ptr& ptr, event_bitmask e, fd_meta_event etype);
void alteration(const continuable_ptr& ptr, event_bitmask e, fd_meta_event etype);
event_bitmask next_bitmask(event_bitmask old, event_bitmask arg, fd_meta_event op) const;
......
......@@ -35,7 +35,7 @@
#include "cppa/ref_counted.hpp"
#include "cppa/intrusive_ptr.hpp"
namespace cppa { namespace network {
namespace cppa { namespace io {
/**
* @brief An abstract output stream interface.
......
......@@ -40,14 +40,14 @@
#include "cppa/ref_counted.hpp"
#include "cppa/primitive_variant.hpp"
#include "cppa/network/acceptor.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/io/acceptor.hpp"
#include "cppa/io/middleman.hpp"
namespace cppa { class actor_addressing; }
namespace cppa { namespace network {
namespace cppa { namespace io {
class continuable_io;
class continuable;
class abstract_middleman;
/**
......@@ -73,7 +73,7 @@ class protocol {
virtual actor_ptr remote_actor(variant_args args) = 0;
virtual actor_ptr remote_actor(io_stream_ptr_pair ioptrs,
virtual actor_ptr remote_actor(stream_ptr_pair ioptrs,
variant_args args ) = 0;
virtual actor_addressing* addressing() = 0;
......@@ -89,25 +89,25 @@ class protocol {
* @brief Convenience member function to be used by children of
* this protocol.
*/
inline void stop_writer(const continuable_io_ptr& ptr);
inline void stop_writer(const continuable_ptr& ptr);
/**
* @brief Convenience member function to be used by children of
* this protocol.
*/
inline void continue_writer(const continuable_io_ptr& ptr);
inline void continue_writer(const continuable_ptr& ptr);
/**
* @brief Convenience member function to be used by children of
* this protocol.
*/
inline void stop_reader(const continuable_io_ptr& ptr);
inline void stop_reader(const continuable_ptr& ptr);
/**
* @brief Convenience member function to be used by children of
* this protocol.
*/
inline void continue_reader(const continuable_io_ptr& ptr);
inline void continue_reader(const continuable_ptr& ptr);
/**
* @brief Returns the parent of this protocol instance.
......@@ -135,19 +135,19 @@ inline void protocol::run_later(T&& what) {
m_parent->run_later(std::forward<T>(what));
}
inline void protocol::stop_writer(const continuable_io_ptr& ptr) {
inline void protocol::stop_writer(const continuable_ptr& ptr) {
m_parent->stop_writer(ptr);
}
inline void protocol::continue_writer(const continuable_io_ptr& ptr) {
inline void protocol::continue_writer(const continuable_ptr& ptr) {
m_parent->continue_writer(ptr);
}
inline void protocol::stop_reader(const continuable_io_ptr& ptr) {
inline void protocol::stop_reader(const continuable_ptr& ptr) {
m_parent->stop_reader(ptr);
}
inline void protocol::continue_reader(const continuable_io_ptr& ptr) {
inline void protocol::continue_reader(const continuable_ptr& ptr) {
m_parent->continue_reader(ptr);
}
......
......@@ -31,20 +31,20 @@
#ifndef CPPA_IO_STREAM_HPP
#define CPPA_IO_STREAM_HPP
#include "cppa/network/input_stream.hpp"
#include "cppa/network/output_stream.hpp"
#include "cppa/io/input_stream.hpp"
#include "cppa/io/output_stream.hpp"
namespace cppa { namespace network {
namespace cppa { namespace io {
/**
* @brief A stream capable of both reading and writing.
*/
class io_stream : public input_stream, public output_stream { };
class stream : public input_stream, public output_stream { };
/**
* @brief An IO stream pointer.
*/
typedef intrusive_ptr<io_stream> io_stream_ptr;
typedef intrusive_ptr<stream> stream_ptr;
} } // namespace cppa::util
......
......@@ -51,7 +51,7 @@ inline detail::actor_registry* get_actor_registry() {
return detail::singleton_manager::get_actor_registry();
}
inline network::middleman* get_middleman() {
inline io::middleman* get_middleman() {
return detail::singleton_manager::get_middleman();
}
......
......@@ -35,7 +35,7 @@
namespace cppa { class deserializer; }
namespace cppa { namespace network { class input_stream; } }
namespace cppa { namespace io { class input_stream; } }
namespace cppa { namespace util {
......@@ -187,7 +187,7 @@ class buffer {
/**
* @brief Appends up to @p remaining() bytes from @p istream to the buffer.
*/
void append_from(network::input_stream* istream);
void append_from(io::input_stream* istream);
/**
* @brief Returns the number of bytes used as minimal allocation unit.
......
......@@ -28,144 +28,158 @@
\******************************************************************************/
#define CPPA_LOG_LEVEL 4
#include <vector>
#include <string>
#include <limits>
#include <memory>
#include <iostream>
#include "cppa/cppa.hpp"
#include "cppa/logging.hpp"
#include "cppa/network/ipv4_acceptor.hpp"
#include "cppa/network/ipv4_io_stream.hpp"
#include "cppa/io/ipv4_acceptor.hpp"
#include "cppa/io/ipv4_io_stream.hpp"
#include "pingpong.pb.h"
#include "cppa/singletons.hpp"
#include "cppa/network/io_actor.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/network/io_service.hpp"
#include "cppa/network/io_actor_backend.hpp"
#include "cppa/io/broker.hpp"
#include "cppa/io/middleman.hpp"
#include "cppa/io/io_handle.hpp"
#include "cppa/io/broker_backend.hpp"
using namespace std;
using namespace cppa;
using namespace cppa::network;
void protobuf_io(io_service* ios) {
partial_function default_bhvr {
on(atom("IO_closed")) >> [=] {
void ping(size_t num_pings) {
auto count = make_shared<size_t>(0);
become (
on(atom("kickoff"), arg_match) >> [=](const actor_ptr& pong) {
send(pong, atom("ping"), 1);
become (
on(atom("pong"), arg_match) >> [=](int value) {
cout << "<- pong " << value << endl;
if (++*count >= num_pings) self->quit();
else reply(atom("ping"), value + 1);
}
);
}
);
}
void pong() {
become (
on(atom("ping"), arg_match) >> [](int value) {
cout << "<- ping " << value << endl;
reply(atom("pong"), value);
}
);
}
void protobuf_io(io_handle* ios, const actor_ptr& buddy) {
self->monitor(buddy);
auto write = [=](const org::libcppa::PingOrPong& p) {
string buf = p.SerializeAsString();
int32_t s = htonl(static_cast<int32_t>(buf.size()));
ios->write(sizeof(int32_t), &s);
ios->write(buf.size(), buf.data());
};
auto default_bhvr = (
on(atom("IO_closed"), arg_match) >> [=](uint32_t) {
cout << "IO_closed" << endl;
self->quit(exit_reason::normal);
self->quit(exit_reason::remote_link_unreachable);
},
on(atom("ping"), arg_match) >> [=](int i) {
org::libcppa::PingOrPong p;
p.mutable_ping()->set_id(i);
write(p);
},
on(atom("pong"), arg_match) >> [=](int i) {
org::libcppa::PingOrPong p;
p.mutable_pong()->set_id(i);
write(p);
},
on(atom("DOWN"), arg_match) >> [=](uint32_t rsn) {
if (self->last_sender() == buddy) self->quit(rsn);
},
others() >> [=] {
cout << "unexpected: " << to_string(self->last_dequeued()) << endl;
}
);
partial_function await_protobuf_data {
on(atom("IO_read"), arg_match) >> [=](uint32_t, const util::buffer& buf) {
org::libcppa::PingOrPong p;
p.ParseFromArray(buf.data(), static_cast<int>(buf.size()));
auto print = [](const char* name, int value) {
cout << name << "{" << value << "}" << endl;
};
if (p.has_ping()) {
send(buddy, atom("ping"), p.ping().id());
}
else if (p.has_pong()) {
send(buddy, atom("pong"), p.pong().id());
}
else {
self->quit(exit_reason::user_defined);
cerr << "neither Pong nor Ping!" << endl;
}
// receive next length prefix
ios->receive_policy(io_handle::exactly, 4);
unbecome();
},
default_bhvr
};
ios->receive_policy(io_service::exactly, 4);
become(
partial_function{
on(atom("IO_read"), arg_match) >> [=](const util::buffer& buf) {
int num_bytes;
memcpy(&num_bytes, buf.data(), 4);
num_bytes = htonl(num_bytes);
if (num_bytes < 0 || num_bytes > (1024 * 1024)) {
self->quit(exit_reason::user_defined);
return;
}
ios->receive_policy(io_service::exactly, (size_t) num_bytes);
become(
partial_function{
on(atom("IO_read"), arg_match) >> [=](const util::buffer& buf) {
org::libcppa::PingOrPong p;
p.ParseFromArray(buf.data(), (int) buf.size());
auto print = [](const char* name, int value) {
cout << name << "{" << value << "}" << endl;
};
if (p.has_ping()) print("Ping", p.ping().id());
else if (p.has_pong()) print("Pong", p.pong().id());
else cerr << "neither Pong nor Ping!" << endl;
self->quit(exit_reason::normal);
}//,default_bhvr
}.or_else(default_bhvr)
);
partial_function await_length_prefix {
on(atom("IO_read"), arg_match) >> [=](uint32_t, const util::buffer& buf) {
int num_bytes;
memcpy(&num_bytes, buf.data(), 4);
num_bytes = htonl(num_bytes);
if (num_bytes < 0 || num_bytes > (1024 * 1024)) {
self->quit(exit_reason::user_defined);
return;
}
}.or_else(default_bhvr)
);
// receive protobuf data
ios->receive_policy(io_handle::exactly, (size_t) num_bytes);
become(keep_behavior, await_protobuf_data);
},
default_bhvr
};
// initial setup
ios->receive_policy(io_handle::exactly, 4);
become(await_length_prefix);
}
/*
class protobuf_io : public io_actor {
enum { await_size, await_data } state;
public:
protobuf_io() : state(await_size) { }
void init() {
io_handle().receive_policy(io_service::exactly, 4);
become (
on(atom("IO_read"), arg_match) >> [=](const util::buffer& buf) {
switch (state) {
case await_size:
int num_bytes;
memcpy(&num_bytes, buf.data(), 4);
num_bytes = htonl(num_bytes);
if (num_bytes < 0 || num_bytes > (1024 * 1024)) {
quit(exit_reason::user_defined);
return;
}
io_handle().receive_policy(io_service::exactly,
static_cast<size_t>(num_bytes));
state = await_data;
break;
case await_data:
org::libcppa::PingOrPong p;
p.ParseFromArray(buf.data(), (int) buf.size());
auto print = [](const char* name, int value) {
cout << name << "{" << value << "}" << endl;
};
if (p.has_ping()) print("Ping", p.ping().id());
else if (p.has_pong()) print("Pong", p.pong().id());
else cerr << "neither Pong nor Ping!" << endl;
quit(exit_reason::normal);
}
},
on(atom("IO_closed")) >> [=] {
cout << "IO_closed" << endl;
quit(exit_reason::normal);
},
others() >> [=] {
cout << "unexpected: " << to_string(last_dequeued()) << endl;
}
);
}
};
*/
int main(int argc, char** argv) {
auto print_exit = [](const actor_ptr& ptr, const std::string& name) {
ptr->attach_functor([=](std::uint32_t reason) {
cout << name << " exited with reason " << reason << endl;
});
};
match(std::vector<string>{argv + 1, argv + argc}) (
on("-s") >> [] {
on("-s") >> [&] {
cout << "run in server mode" << endl;
auto ack = network::ipv4_acceptor::create(4242);
auto p = ack->accept_connection();
//spawn_io<protobuf_io>(p.first, p.second);
spawn_io(protobuf_io, std::move(p.first), std::move(p.second));
await_all_others_done();
std::string hi = "hello";
auto po = spawn(pong);
auto ack = io::ipv4_acceptor::create(4242);
for (;;) {
auto p = ack->accept_connection();
//spawn_io<protobuf_io>(p.first, p.second);
auto s = spawn_io(protobuf_io, std::move(p.first), std::move(p.second), po);
print_exit(s, "io actor");
print_exit(po, "pong");
}
},
on_arg_match >> [](const string& host, const string& port_str) {
on_arg_match >> [&](const string& host, const string& port_str) {
auto port = static_cast<uint16_t>(stoul(port_str));
auto serv = network::ipv4_io_stream::connect_to(host.c_str(), port);
org::libcppa::PingOrPong p;
p.mutable_ping()->set_id(numeric_limits<int32_t>::max());
string buf = p.SerializeAsString();
int32_t s = htonl(static_cast<int32_t>(buf.size()));
serv->write(&s, 4);
serv->write(buf.data(), buf.size());
cout << "run in client mode" << endl;
auto io = io::ipv4_io_stream::connect_to(host.c_str(), port);
auto pi = spawn(ping, 20);
auto pr = spawn_io(protobuf_io, io, io, pi);
send_as(pr, pi, atom("kickoff"), pr);
print_exit(pr, "io actor");
print_exit(pi, "ping");
}
);
await_all_others_done();
shutdown();
}
......@@ -37,7 +37,7 @@
#include "cppa/scheduler.hpp"
#include "cppa/actor_proxy.hpp"
#include "cppa/exit_reason.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/io/middleman.hpp"
#include "cppa/detail/types_array.hpp"
#include "cppa/detail/singleton_manager.hpp"
......
......@@ -28,22 +28,23 @@
\******************************************************************************/
#include "cppa/cppa.hpp"
#include "cppa/singletons.hpp"
#include "cppa/detail/actor_registry.hpp"
#include "cppa/network/io_actor.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/network/io_actor_backend.hpp"
#include "cppa/io/broker.hpp"
#include "cppa/io/middleman.hpp"
#include "cppa/io/broker_backend.hpp"
#include "cppa/detail/sync_request_bouncer.hpp"
namespace cppa { namespace network {
namespace cppa { namespace io {
class io_actor_continuation {
class broker_continuation {
public:
io_actor_continuation(io_actor_ptr ptr, mailbox_element* elem)
broker_continuation(broker_ptr ptr, mailbox_element* elem)
: m_self(std::move(ptr)), m_elem(elem) { }
inline void operator()() const {
......@@ -52,21 +53,26 @@ class io_actor_continuation {
private:
io_actor_ptr m_self;
broker_ptr m_self;
mailbox_element* m_elem;
};
class default_io_actor_impl : public io_actor {
class default_broker_impl : public broker {
public:
typedef std::function<void (io_service*)> function_type;
typedef std::function<void (io_handle*)> function_type;
default_io_actor_impl(function_type&& fun) : m_fun(std::move(fun)) { }
default_broker_impl(function_type&& fun) : m_fun(std::move(fun)) { }
void init() override {
m_fun(&io_handle());
enqueue(nullptr, make_any_tuple(atom("INITMSG")));
become(
on(atom("INITMSG")) >> [=] {
m_fun(&io());
}
);
}
private:
......@@ -75,7 +81,7 @@ class default_io_actor_impl : public io_actor {
};
void io_actor::invoke_message(mailbox_element* elem) {
void broker::invoke_message(mailbox_element* elem) {
if (exit_reason() != exit_reason::not_exited) {
if (elem->mid.valid()) {
detail::sync_request_bouncer srb{exit_reason()};
......@@ -106,30 +112,30 @@ void io_actor::invoke_message(mailbox_element* elem) {
}
}
void io_actor::invoke_message(any_tuple msg) {
void broker::invoke_message(any_tuple msg) {
invoke_message(mailbox_element::create(this, std::move(msg)));
}
void io_actor::enqueue(const message_header& hdr, any_tuple msg) {
void broker::enqueue(const message_header& hdr, any_tuple msg) {
auto e = mailbox_element::create(hdr, std::move(msg));
get_middleman()->run_later(io_actor_continuation{this, e});
get_middleman()->run_later(broker_continuation{this, e});
}
bool io_actor::initialized() const {
bool broker::initialized() const {
return true;
}
void io_actor::quit(std::uint32_t reason) {
void broker::quit(std::uint32_t reason) {
cleanup(reason);
m_parent->handle_disconnect();
}
io_service& io_actor::io_handle() {
io_handle& broker::io() {
return *m_parent;
}
intrusive_ptr<io_actor> io_actor::from(std::function<void (io_service*)> fun) {
return make_counted<default_io_actor_impl>(std::move(fun));
intrusive_ptr<broker> broker::from(std::function<void (io_handle*)> fun) {
return make_counted<default_broker_impl>(std::move(fun));
}
} } // namespace cppa::network
......@@ -30,70 +30,60 @@
#include "cppa/singletons.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/network/io_actor_backend.hpp"
#include "cppa/io/middleman.hpp"
#include "cppa/io/broker_backend.hpp"
#include "cppa/detail/actor_registry.hpp"
namespace cppa { namespace network {
namespace cppa { namespace io {
io_actor_backend::io_actor_backend(input_stream_ptr in,
broker_backend::broker_backend(input_stream_ptr in,
output_stream_ptr out,
io_actor_ptr ptr)
broker_ptr ptr)
: super(get_middleman(), in->read_handle(), std::move(out))
, m_in(in), m_self(ptr) {
, m_dirty(false), m_policy(at_least)
, m_policy_buffer_size(0)
, m_in(in), m_self(ptr)
, m_read(atom("IO_read"), static_cast<uint32_t>(in->read_handle())) {
m_self->m_parent = this;
get_ref<0>(m_read) = atom("IO_read");
get_ref<1>(m_read).final_size(default_max_buffer_size);
m_dirty = false;
m_policy = at_least;
m_policy_buffer_size = 0;
get_ref<2>(m_read).final_size(default_max_buffer_size);
}
io_actor_backend::~io_actor_backend() {
handle_disconnect();
}
void io_actor_backend::init() {
void broker_backend::init() {
get_actor_registry()->inc_running();
auto selfptr = m_self.get();
scoped_self_setter sss{selfptr};
selfptr->init();
}
void io_actor_backend::handle_disconnect() {
if (m_self) {
auto ms = m_self;
m_self.reset();
bool dec_count = ms->exit_reason() == exit_reason::not_exited;
CPPA_LOG_DEBUG("became disconnected");
if (ms->exit_reason() == exit_reason::not_exited) {
ms->invoke_message(make_any_tuple(atom("IO_closed")));
}
get_middleman()->stop_reader(this);
if (dec_count) {
if (ms->exit_reason() != exit_reason::not_exited) {
get_actor_registry()->dec_running();
}
}
void broker_backend::handle_disconnect() {
if (m_self == nullptr) return;
auto ms = m_self;
m_self.reset();
CPPA_LOG_DEBUG("became disconnected");
if (ms->exit_reason() == exit_reason::not_exited) {
auto msg = make_any_tuple(atom("IO_closed"),
static_cast<uint32_t>(m_in->read_handle()));
ms->invoke_message(std::move(msg));
}
get_middleman()->stop_reader(this);
}
void io_actor_backend::io_failed() {
void broker_backend::io_failed() {
handle_disconnect();
}
void io_actor_backend::receive_policy(policy_flag policy, size_t buffer_size) {
void broker_backend::receive_policy(policy_flag policy, size_t buffer_size) {
CPPA_LOG_TRACE(CPPA_ARG(policy) << ", " << CPPA_ARG(buffer_size));
m_dirty = true;
m_policy = policy;
m_policy_buffer_size = buffer_size;
}
continue_reading_result io_actor_backend::continue_reading() {
continue_reading_result broker_backend::continue_reading() {
CPPA_LOG_TRACE("");
for (;;) {
auto& buf = get_ref<1>(m_read);
auto& buf = get_ref<2>(m_read);
if (m_dirty) {
m_dirty = false;
if (m_policy == at_most || m_policy == exactly) {
......@@ -119,21 +109,21 @@ continue_reading_result io_actor_backend::continue_reading() {
m_self->invoke_message(m_read);
CPPA_LOG_INFO_IF(!m_read.vals()->unique(), "buffer became detached");
if (m_self == nullptr) {
// io_actor::quit() calls handle_disconnect, which sets
// broker::quit() calls handle_disconnect, which sets
// m_self to nullptr
return read_closed;
}
get_ref<1>(m_read).clear();
get_ref<2>(m_read).clear();
}
}
}
void io_actor_backend::close() {
void broker_backend::close() {
CPPA_LOG_DEBUG("");
get_middleman()->stop_reader(this);
}
void io_actor_backend::write(size_t num_bytes, const void* data) {
void broker_backend::write(size_t num_bytes, const void* data) {
super::write(num_bytes, data);
}
......
......@@ -33,7 +33,7 @@
#include <utility>
#include "cppa/util/buffer.hpp"
#include "cppa/network/input_stream.hpp"
#include "cppa/io/input_stream.hpp"
namespace cppa { namespace util {
......@@ -162,7 +162,7 @@ void buffer::write(buffer&& other, buffer_write_policy wp) {
other.clear();
}
void buffer::append_from(network::input_stream* istream) {
void buffer::append_from(io::input_stream* istream) {
CPPA_REQUIRE(remaining() > 0);
auto num_bytes = istream->read_some(wr_ptr(), remaining());
if (num_bytes > 0) {
......
......@@ -32,10 +32,10 @@
#include "cppa/to_string.hpp"
#include "cppa/singletons.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/network/buffered_writer.hpp"
#include "cppa/io/middleman.hpp"
#include "cppa/io/buffered_writer.hpp"
namespace cppa { namespace network {
namespace cppa { namespace io {
buffered_writer::buffered_writer(middleman* pptr, native_socket_type rfd, output_stream_ptr out)
: super(rfd, out->write_handle()), m_middleman(pptr)
......
......@@ -28,18 +28,18 @@
\******************************************************************************/
#include "cppa/network/continuable_io.hpp"
#include "cppa/io/continuable.hpp"
namespace cppa { namespace network {
namespace cppa { namespace io {
continuable_io::continuable_io(native_socket_type rd, native_socket_type wr)
continuable::continuable(native_socket_type rd, native_socket_type wr)
: m_rd(rd), m_wr(wr) { }
continue_reading_result continuable_io::continue_reading() {
continue_reading_result continuable::continue_reading() {
return read_closed;
}
continue_writing_result continuable_io::continue_writing() {
continue_writing_result continuable::continue_writing() {
return write_closed;
}
......
......@@ -36,15 +36,15 @@
#include "cppa/deserializer.hpp"
#include "cppa/primitive_variant.hpp"
#include "cppa/network/default_actor_proxy.hpp"
#include "cppa/network/default_actor_addressing.hpp"
#include "cppa/io/default_actor_proxy.hpp"
#include "cppa/io/default_actor_addressing.hpp"
#include "cppa/detail/actor_registry.hpp"
#include "cppa/detail/singleton_manager.hpp"
using namespace std;
namespace cppa { namespace network {
namespace cppa { namespace io {
default_actor_addressing::default_actor_addressing(default_protocol* parent)
: m_parent(parent), m_pinf(process_information::get()) { }
......
......@@ -31,8 +31,8 @@
#include "cppa/to_string.hpp"
#include "cppa/logging.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/network/default_actor_proxy.hpp"
#include "cppa/io/middleman.hpp"
#include "cppa/io/default_actor_proxy.hpp"
#include "cppa/detail/memory.hpp"
#include "cppa/detail/singleton_manager.hpp"
......@@ -40,7 +40,7 @@
using namespace std;
namespace cppa { namespace network {
namespace cppa { namespace io {
inline sync_request_info* new_req_info(actor_ptr sptr, message_id id) {
return detail::memory::create<sync_request_info>(std::move(sptr), id);
......@@ -65,7 +65,7 @@ default_actor_proxy::~default_actor_proxy() {
CPPA_LOG_INFO(CPPA_ARG(m_id) << ", " << CPPA_TSARG(m_pinf)
<< ", protocol = " << detail::demangle(typeid(*m_parent)));
proto->run_later([aid, node, proto] {
CPPA_LOGC_TRACE("cppa::network::default_actor_proxy",
CPPA_LOGC_TRACE("cppa::io::default_actor_proxy",
"~default_actor_proxy$run_later",
"node = " << to_string(*node) << ", aid " << aid
<< ", proto = " << to_string(proto->identifier()));
......@@ -107,7 +107,7 @@ void default_actor_proxy::forward_msg(const message_header& hdr, any_tuple msg)
case intrusive::queue_closed: {
auto rsn = exit_reason();
m_parent->run_later([rsn, hdr] {
CPPA_LOGC_TRACE("cppa::network::default_actor_proxy",
CPPA_LOGC_TRACE("cppa::io::default_actor_proxy",
"forward_msg$bouncer",
"bounce message for reason " << rsn);
detail::sync_request_bouncer f{rsn};
......@@ -121,7 +121,7 @@ void default_actor_proxy::forward_msg(const message_header& hdr, any_tuple msg)
auto node = m_pinf;
auto proto = m_parent;
m_parent->run_later([hdr, msg, node, proto] {
CPPA_LOGC_TRACE("cppa::network::default_actor_proxy",
CPPA_LOGC_TRACE("cppa::io::default_actor_proxy",
"forward_msg$forwarder",
"");
proto->enqueue(*node, hdr, msg);
......@@ -139,7 +139,7 @@ void default_actor_proxy::enqueue(const message_header& hdr, any_tuple msg) {
intrusive_ptr<default_actor_proxy> _this{this};
auto reason = msg.get_as<uint32_t>(1);
m_parent->run_later([_this, reason] {
CPPA_LOGC_TRACE("cppa::network::default_actor_proxy",
CPPA_LOGC_TRACE("cppa::io::default_actor_proxy",
"enqueue$kill_proxy_helper",
"KILL_PROXY with exit reason " << reason);
_this->cleanup(reason);
......
......@@ -47,14 +47,14 @@
#include "cppa/detail/actor_registry.hpp"
#include "cppa/detail/singleton_manager.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/network/default_peer.hpp"
#include "cppa/io/middleman.hpp"
#include "cppa/io/default_peer.hpp"
#include "cppa/message_header.hpp"
#include "cppa/network/default_protocol.hpp"
#include "cppa/io/default_protocol.hpp"
using namespace std;
namespace cppa { namespace network {
namespace cppa { namespace io {
default_peer::default_peer(default_protocol* parent,
const input_stream_ptr& in,
......@@ -215,7 +215,7 @@ void default_peer::monitor(const actor_ptr&,
default_protocol* proto = m_parent;
entry.first->attach_functor([=](uint32_t reason) {
proto->run_later([=] {
CPPA_LOGC_TRACE("cppa::network::default_peer",
CPPA_LOGC_TRACE("cppa::io::default_peer",
"monitor$kill_proxy_helper",
"reason = " << reason);
auto p = proto->get_peer(*node);
......
......@@ -35,15 +35,15 @@
#include "cppa/to_string.hpp"
#include "cppa/process_information.hpp"
#include "cppa/network/default_protocol.hpp"
#include "cppa/network/default_peer.hpp"
#include "cppa/network/default_peer_acceptor.hpp"
#include "cppa/io/default_protocol.hpp"
#include "cppa/io/default_peer.hpp"
#include "cppa/io/default_peer_acceptor.hpp"
#include "cppa/detail/demangle.hpp"
using namespace std;
namespace cppa { namespace network {
namespace cppa { namespace io {
default_peer_acceptor::default_peer_acceptor(default_protocol* parent,
acceptor_uptr aur,
......@@ -53,7 +53,7 @@ default_peer_acceptor::default_peer_acceptor(default_protocol* parent,
continue_reading_result default_peer_acceptor::continue_reading() {
CPPA_LOG_TRACE("");
for (;;) {
option<io_stream_ptr_pair> opt;
option<stream_ptr_pair> opt;
try { opt = m_ptr->try_accept_connection(); }
catch (exception& e) {
CPPA_LOG_ERROR(to_verbose_string(e));
......
......@@ -36,12 +36,12 @@
#include "cppa/to_string.hpp"
#include "cppa/singletons.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/network/default_peer.hpp"
#include "cppa/network/ipv4_acceptor.hpp"
#include "cppa/network/ipv4_io_stream.hpp"
#include "cppa/network/default_protocol.hpp"
#include "cppa/network/default_peer_acceptor.hpp"
#include "cppa/io/middleman.hpp"
#include "cppa/io/default_peer.hpp"
#include "cppa/io/ipv4_acceptor.hpp"
#include "cppa/io/ipv4_io_stream.hpp"
#include "cppa/io/default_protocol.hpp"
#include "cppa/io/default_peer_acceptor.hpp"
#include "cppa/detail/actor_registry.hpp"
#include "cppa/detail/singleton_manager.hpp"
......@@ -51,7 +51,7 @@
using namespace std;
using namespace cppa::detail;
namespace cppa { namespace network {
namespace cppa { namespace io {
default_protocol::default_protocol(middleman* parent)
: super(parent), m_addressing(this) { }
......@@ -93,7 +93,7 @@ void default_protocol::publish(const actor_ptr& whom,
default_protocol* proto = this;
auto impl = make_counted<default_peer_acceptor>(this, move(ptr), whom);
run_later([=] {
CPPA_LOGC_TRACE("cppa::network::default_protocol",
CPPA_LOGC_TRACE("cppa::io::default_protocol",
"publish$add_acceptor", "");
proto->m_acceptors[whom].push_back(impl);
proto->continue_reader(impl.get());
......@@ -104,7 +104,7 @@ void default_protocol::unpublish(const actor_ptr& whom) {
CPPA_LOG_TRACE("whom = " << to_string(whom));
default_protocol* proto = this;
run_later([=] {
CPPA_LOGC_TRACE("cppa::network::default_protocol",
CPPA_LOGC_TRACE("cppa::io::default_protocol",
"unpublish$remove_acceptors", "");
auto& acceptors = m_acceptors[whom];
for (auto& ptr : acceptors) proto->stop_reader(ptr.get());
......@@ -167,12 +167,12 @@ actor_ptr default_protocol::remote_actor(variant_args args) {
auto port = get<uint16_t>(*i++);
auto& host = get<string>(*i);
auto io = ipv4_io_stream::connect_to(host.c_str(), port);
return remote_actor(io_stream_ptr_pair(io, io), {});
return remote_actor(stream_ptr_pair(io, io), {});
}
struct remote_actor_result { remote_actor_result* next; actor_ptr value; };
actor_ptr default_protocol::remote_actor(io_stream_ptr_pair io,
actor_ptr default_protocol::remote_actor(stream_ptr_pair io,
variant_args args ) {
CPPA_LOG_TRACE("io = {" << io.first.get() << ", " << io.second.get() << "}, "
<< "args.size() = " << args.size());
......@@ -202,7 +202,7 @@ actor_ptr default_protocol::remote_actor(io_stream_ptr_pair io,
default_protocol* proto = this;
intrusive::blocking_single_reader_queue<remote_actor_result> q;
run_later([proto, io, pinfptr, remote_aid, &q] {
CPPA_LOGC_TRACE("cppa::network::default_protocol",
CPPA_LOGC_TRACE("cppa::io::default_protocol",
"remote_actor$create_connection", "");
auto pp = proto->get_peer(*pinfptr);
CPPA_LOGF_INFO_IF(pp, "connection already exists (re-use old one)");
......
......@@ -42,7 +42,7 @@
#include "cppa/deserializer.hpp"
#include "cppa/event_based_actor.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/io/middleman.hpp"
#include "cppa/detail/types_array.hpp"
#include "cppa/detail/group_manager.hpp"
#include "cppa/message_header.hpp"
......
......@@ -28,10 +28,10 @@
\******************************************************************************/
#include "cppa/network/io_service.hpp"
#include "cppa/io/io_handle.hpp"
namespace cppa { namespace network {
namespace cppa { namespace io {
io_service::~io_service() { }
io_handle::~io_handle() { }
} } // namespace cppa::network
......@@ -34,9 +34,9 @@
#include <iostream>
#include "cppa/exception.hpp"
#include "cppa/network/io_stream.hpp"
#include "cppa/network/ipv4_acceptor.hpp"
#include "cppa/network/ipv4_io_stream.hpp"
#include "cppa/io/stream.hpp"
#include "cppa/io/ipv4_acceptor.hpp"
#include "cppa/io/ipv4_io_stream.hpp"
#include "cppa/detail/fd_util.hpp"
......@@ -51,7 +51,7 @@
# include <netinet/tcp.h>
#endif
namespace cppa { namespace network {
namespace cppa { namespace io {
using namespace ::cppa::detail::fd_util;
......@@ -76,7 +76,7 @@ struct socket_guard {
};
bool accept_impl(io_stream_ptr_pair& result,
bool accept_impl(stream_ptr_pair& result,
native_socket_type fd,
bool nonblocking) {
sockaddr addr;
......@@ -91,7 +91,7 @@ bool accept_impl(io_stream_ptr_pair& result,
}
throw_io_failure("accept failed");
}
io_stream_ptr ptr(ipv4_io_stream::from_native_socket(sfd));
stream_ptr ptr(ipv4_io_stream::from_native_socket(sfd));
result.first = ptr;
result.second = ptr;
return true;
......@@ -147,22 +147,22 @@ native_socket_type ipv4_acceptor::file_handle() const {
return m_fd;
}
io_stream_ptr_pair ipv4_acceptor::accept_connection() {
stream_ptr_pair ipv4_acceptor::accept_connection() {
if (m_is_nonblocking) {
nonblocking(m_fd, false);
m_is_nonblocking = false;
}
io_stream_ptr_pair result;
stream_ptr_pair result;
accept_impl(result, m_fd, m_is_nonblocking);
return result;
}
option<io_stream_ptr_pair> ipv4_acceptor::try_accept_connection() {
option<stream_ptr_pair> ipv4_acceptor::try_accept_connection() {
if (!m_is_nonblocking) {
nonblocking(m_fd, true);
m_is_nonblocking = true;
}
io_stream_ptr_pair result;
stream_ptr_pair result;
if (accept_impl(result, m_fd, m_is_nonblocking)) {
return result;
}
......
......@@ -36,7 +36,7 @@
#include "cppa/logging.hpp"
#include "cppa/exception.hpp"
#include "cppa/detail/fd_util.hpp"
#include "cppa/network/ipv4_io_stream.hpp"
#include "cppa/io/ipv4_io_stream.hpp"
#ifdef CPPA_WINDOWS
#else
......@@ -48,7 +48,7 @@
# include <netinet/tcp.h>
#endif
namespace cppa { namespace network {
namespace cppa { namespace io {
using namespace ::cppa::detail::fd_util;
......@@ -115,13 +115,13 @@ size_t ipv4_io_stream::write_some(const void* buf, size_t len) {
return static_cast<size_t>(send_result);
}
network::io_stream_ptr ipv4_io_stream::from_native_socket(native_socket_type fd) {
io::stream_ptr ipv4_io_stream::from_native_socket(native_socket_type fd) {
tcp_nodelay(fd, true);
nonblocking(fd, true);
return new ipv4_io_stream(fd);
}
network::io_stream_ptr ipv4_io_stream::connect_to(const char* host,
io::stream_ptr ipv4_io_stream::connect_to(const char* host,
std::uint16_t port) {
CPPA_LOGF_INFO("try to connect to " << host << " on port " << port);
struct sockaddr_in serv_addr;
......
......@@ -51,13 +51,13 @@
#include "cppa/util/buffer.hpp"
#include "cppa/network/protocol.hpp"
#include "cppa/network/acceptor.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/network/input_stream.hpp"
#include "cppa/network/output_stream.hpp"
#include "cppa/network/default_protocol.hpp"
#include "cppa/network/middleman_event_handler.hpp"
#include "cppa/io/protocol.hpp"
#include "cppa/io/acceptor.hpp"
#include "cppa/io/middleman.hpp"
#include "cppa/io/input_stream.hpp"
#include "cppa/io/output_stream.hpp"
#include "cppa/io/default_protocol.hpp"
#include "cppa/io/middleman_event_handler.hpp"
#include "cppa/detail/fd_util.hpp"
#include "cppa/detail/actor_registry.hpp"
......@@ -66,7 +66,7 @@
using namespace std;
namespace cppa { namespace network {
namespace cppa { namespace io {
class middleman_event {
......@@ -111,27 +111,27 @@ class middleman_impl {
static_cast<void>(write(m_pipe_write, &dummy, sizeof(dummy)));
}
void continue_writer(const continuable_io_ptr& ptr) {
void continue_writer(const continuable_ptr& ptr) {
CPPA_LOG_TRACE("ptr = " << ptr.get());
m_handler->add_later(ptr, event::write);
}
void stop_writer(const continuable_io_ptr& ptr) {
void stop_writer(const continuable_ptr& ptr) {
CPPA_LOG_TRACE("ptr = " << ptr.get());
m_handler->erase_later(ptr, event::write);
}
void continue_reader(const continuable_io_ptr& ptr) {
void continue_reader(const continuable_ptr& ptr) {
CPPA_LOG_TRACE("ptr = " << ptr.get());
m_readers.push_back(ptr);
m_handler->add_later(ptr, event::read);
}
void stop_reader(const continuable_io_ptr& ptr) {
void stop_reader(const continuable_ptr& ptr) {
CPPA_LOG_TRACE("ptr = " << ptr.get());
m_handler->erase_later(ptr, event::read);
auto last = m_readers.end();
auto i = find_if(m_readers.begin(), last, [&](const continuable_io_ptr& lhs) {
auto i = find_if(m_readers.begin(), last, [&](const continuable_ptr& lhs) {
return lhs == ptr;
});
if (i != last) m_readers.erase(i);
......@@ -166,7 +166,7 @@ class middleman_impl {
inline bool done() const { return m_done; }
bool m_done;
std::vector<continuable_io_ptr> m_readers;
std::vector<continuable_ptr> m_readers;
middleman_event_handler& handler();
......@@ -190,9 +190,9 @@ middleman* middleman::create_singleton() {
return ptr;
}
class middleman_overseer : public continuable_io {
class middleman_overseer : public continuable {
typedef continuable_io super;
typedef continuable super;
public:
......@@ -254,19 +254,19 @@ void middleman::run_later(std::function<void()> fun) {
m_impl->run_later(std::move(fun));
}
void middleman::continue_writer(const continuable_io_ptr& ptr) {
void middleman::continue_writer(const continuable_ptr& ptr) {
m_impl->continue_writer(ptr);
}
void middleman::stop_writer(const continuable_io_ptr& ptr) {
void middleman::stop_writer(const continuable_ptr& ptr) {
m_impl->stop_writer(ptr);
}
void middleman::continue_reader(const continuable_io_ptr& ptr) {
void middleman::continue_reader(const continuable_ptr& ptr) {
m_impl->continue_reader(ptr);
}
void middleman::stop_reader(const continuable_io_ptr& ptr) {
void middleman::stop_reader(const continuable_ptr& ptr) {
m_impl->stop_reader(ptr);
}
......@@ -284,7 +284,7 @@ void middleman_loop(middleman_impl* impl) {
impl->continue_reader(make_counted<middleman_overseer>(impl->m_pipe_read, impl->m_queue));
handler->update();
while (!impl->done()) {
handler->poll([&](event_bitmask mask, continuable_io* io) {
handler->poll([&](event_bitmask mask, continuable* io) {
switch (mask) {
default: CPPA_CRITICAL("invalid event");
case event::none: break;
......@@ -335,7 +335,7 @@ void middleman_loop(middleman_impl* impl) {
CPPA_LOGF_DEBUG_IF(handler->num_sockets() == 0,
"nothing to flush, no writer left");
while (handler->num_sockets() > 0) {
handler->poll([&](event_bitmask mask, continuable_io* io) {
handler->poll([&](event_bitmask mask, continuable* io) {
switch (mask) {
case event::write:
switch (io->continue_writing()) {
......
......@@ -28,9 +28,9 @@
\******************************************************************************/
#include "cppa/network/middleman_event_handler.hpp"
#include "cppa/io/middleman_event_handler.hpp"
namespace cppa { namespace network {
namespace cppa { namespace io {
inline std::string eb2str(event_bitmask e) {
switch (e) {
......@@ -47,7 +47,7 @@ middleman_event_handler::middleman_event_handler() { }
middleman_event_handler::~middleman_event_handler() { }
void middleman_event_handler::alteration(const continuable_io_ptr& ptr,
void middleman_event_handler::alteration(const continuable_ptr& ptr,
event_bitmask e,
fd_meta_event etype) {
native_socket_type fd;
......@@ -78,12 +78,12 @@ void middleman_event_handler::alteration(const continuable_io_ptr& ptr,
m_alterations.emplace_back(fd_meta_info(fd, ptr, e), etype);
}
void middleman_event_handler::add_later(const continuable_io_ptr& ptr, event_bitmask e) {
void middleman_event_handler::add_later(const continuable_ptr& ptr, event_bitmask e) {
CPPA_LOG_TRACE("ptr = " << ptr.get() << ", e = " << eb2str(e));
alteration(ptr, e, fd_meta_event::add);
}
void middleman_event_handler::erase_later(const continuable_io_ptr& ptr, event_bitmask e) {
void middleman_event_handler::erase_later(const continuable_ptr& ptr, event_bitmask e) {
CPPA_LOG_TRACE("ptr = " << ptr.get() << ", e = " << eb2str(e));
alteration(ptr, e, fd_meta_event::erase);
}
......
......@@ -34,9 +34,9 @@
#include <string.h>
#include <sys/epoll.h>
#include "cppa/network/middleman_event_handler.hpp"
#include "cppa/io/middleman_event_handler.hpp"
namespace cppa { namespace network {
namespace cppa { namespace io {
namespace {
......@@ -94,7 +94,7 @@ class middleman_event_handler_impl : public middleman_event_handler {
auto eb = from_int_bitmask<input_event,
output_event,
error_event>(iter->events);
auto ptr = reinterpret_cast<continuable_io*>(iter->data.ptr);
auto ptr = reinterpret_cast<continuable*>(iter->data.ptr);
CPPA_REQUIRE(eb != event::none);
m_events.emplace_back(eb, ptr);
}
......@@ -104,7 +104,7 @@ class middleman_event_handler_impl : public middleman_event_handler {
native_socket_type fd,
event_bitmask,
event_bitmask new_bitmask,
continuable_io* ptr) {
continuable* ptr) {
int operation;
epoll_event ee;
ee.data.ptr = ptr;
......
......@@ -29,13 +29,13 @@
#include <poll.h>
#include "cppa/network/middleman_event_handler.hpp"
#include "cppa/io/middleman_event_handler.hpp"
#ifndef POLLRDHUP
#define POLLRDHUP POLLHUP
#endif
namespace cppa { namespace network {
namespace cppa { namespace io {
namespace {
......@@ -106,7 +106,7 @@ class middleman_event_handler_impl : public middleman_event_handler {
native_socket_type fd,
event_bitmask,
event_bitmask new_bitmask,
continuable_io*) {
continuable*) {
auto last = m_pollset.end();
auto iter = std::lower_bound(m_pollset.begin(), last, fd, pollfd_less);
switch (me) {
......
......@@ -29,10 +29,10 @@
#include "cppa/logging.hpp"
#include "cppa/network/protocol.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/io/protocol.hpp"
#include "cppa/io/middleman.hpp"
namespace cppa { namespace network {
namespace cppa { namespace io {
protocol::protocol(middleman* parent) : m_parent(parent) {
CPPA_REQUIRE(parent != nullptr);
......
......@@ -38,7 +38,7 @@
#include "cppa/local_actor.hpp"
#include "cppa/thread_mapped_actor.hpp"
#include "cppa/network/middleman.hpp"
#include "cppa/io/middleman.hpp"
#include "cppa/detail/empty_tuple.hpp"
#include "cppa/detail/group_manager.hpp"
......@@ -65,7 +65,7 @@ namespace {
std::atomic<opencl::opencl_metainfo*> s_opencl_metainfo;
std::atomic<uniform_type_info_map*> s_uniform_type_info_map;
std::atomic<network::middleman*> s_middleman;
std::atomic<io::middleman*> s_middleman;
std::atomic<actor_registry*> s_actor_registry;
std::atomic<group_manager*> s_group_manager;
std::atomic<empty_tuple*> s_empty_tuple;
......@@ -142,7 +142,7 @@ bool singleton_manager::set_scheduler(scheduler* ptr) {
}
}
network::middleman* singleton_manager::get_middleman() {
io::middleman* singleton_manager::get_middleman() {
return lazy_get(s_middleman);
}
......
......@@ -44,7 +44,7 @@
#include "cppa/primitive_variant.hpp"
#include "cppa/uniform_type_info.hpp"
#include "cppa/network/default_actor_addressing.hpp"
#include "cppa/io/default_actor_addressing.hpp"
using namespace std;
......@@ -64,7 +64,7 @@ class string_serializer : public serializer {
typedef serializer super;
ostream& out;
network::default_actor_addressing m_addressing;
io::default_actor_addressing m_addressing;
struct pt_writer {
......@@ -202,7 +202,7 @@ class string_deserializer : public deserializer {
//size_t m_obj_count;
stack<bool> m_obj_had_left_parenthesis;
stack<string> m_open_objects;
network::default_actor_addressing m_addressing;
io::default_actor_addressing m_addressing;
void skip_space_and_comma() {
while (*m_pos == ' ' || *m_pos == ',') ++m_pos;
......
......@@ -50,16 +50,16 @@
#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/io/acceptor.hpp"
#include "cppa/io/protocol.hpp"
#include "cppa/io/middleman.hpp"
#include "cppa/io/ipv4_acceptor.hpp"
#include "cppa/io/ipv4_io_stream.hpp"
namespace cppa {
using namespace detail;
using namespace network;
using namespace io;
namespace { protocol* proto() {
return get_middleman()->get_protocol();
......@@ -69,7 +69,7 @@ void publish(actor_ptr whom, std::unique_ptr<acceptor> aptr) {
proto()->publish(whom, move(aptr), {});
}
actor_ptr remote_actor(io_stream_ptr_pair io) {
actor_ptr remote_actor(stream_ptr_pair io) {
return proto()->remote_actor(io, {});
}
......
......@@ -47,7 +47,7 @@
#include "cppa/util/type_traits.hpp"
#include "cppa/util/abstract_uniform_type_info.hpp"
#include "cppa/network/default_actor_addressing.hpp"
#include "cppa/io/default_actor_addressing.hpp"
#include "cppa/detail/object_array.hpp"
#include "cppa/detail/type_to_ptype.hpp"
......@@ -156,7 +156,7 @@ int main() {
announce(typeid(raw_struct), new raw_struct_type_info);
network::default_actor_addressing addressing;
io::default_actor_addressing addressing;
cout << "process id: " << to_string(process_information::get()) << endl;
......
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