Commit db667857 authored by Dominik Charousset's avatar Dominik Charousset

extended broker and added simple(r) broker example

this patch adds the member functions `add_connection`,
`add_tcp_connection`, `add_acceptor`, and `add_tcp_acceptor`
to the broker class and adds the simple_broker example to
showcase the broker API without Google Protobuf overhead
parent e1d3ca68
...@@ -147,9 +147,10 @@ set(LIBCPPA_SRC ...@@ -147,9 +147,10 @@ set(LIBCPPA_SRC
src/actor_addr.cpp src/actor_addr.cpp
src/actor_companion.cpp src/actor_companion.cpp
src/actor_namespace.cpp src/actor_namespace.cpp
src/actor_ostream.cpp
src/actor_proxy.cpp
src/actor_proxy.cpp src/actor_proxy.cpp
src/actor_registry.cpp src/actor_registry.cpp
src/actor_ostream.cpp
src/algorithm.cpp src/algorithm.cpp
src/any_tuple.cpp src/any_tuple.cpp
src/atom.cpp src/atom.cpp
...@@ -158,9 +159,9 @@ set(LIBCPPA_SRC ...@@ -158,9 +159,9 @@ set(LIBCPPA_SRC
src/behavior_stack.cpp src/behavior_stack.cpp
src/binary_deserializer.cpp src/binary_deserializer.cpp
src/binary_serializer.cpp src/binary_serializer.cpp
src/blocking_actor.cpp
src/broker.cpp src/broker.cpp
src/buffer.cpp src/buffer.cpp
src/blocking_actor.cpp
src/channel.cpp src/channel.cpp
src/context_switching_resume.cpp src/context_switching_resume.cpp
src/continuable.cpp src/continuable.cpp
...@@ -168,9 +169,6 @@ set(LIBCPPA_SRC ...@@ -168,9 +169,6 @@ set(LIBCPPA_SRC
src/cs_thread.cpp src/cs_thread.cpp
src/decorated_tuple.cpp src/decorated_tuple.cpp
src/default_message_queue.cpp src/default_message_queue.cpp
src/actor_proxy.cpp
src/peer.cpp
src/peer_acceptor.cpp
src/demangle.cpp src/demangle.cpp
src/deserializer.cpp src/deserializer.cpp
src/duration.cpp src/duration.cpp
...@@ -181,13 +179,11 @@ set(LIBCPPA_SRC ...@@ -181,13 +179,11 @@ set(LIBCPPA_SRC
src/fd_util.cpp src/fd_util.cpp
src/functor_based_actor.cpp src/functor_based_actor.cpp
src/functor_based_blocking_actor.cpp src/functor_based_blocking_actor.cpp
src/get_root_uuid.cpp
src/get_mac_addresses.cpp src/get_mac_addresses.cpp
src/get_root_uuid.cpp
src/group.cpp src/group.cpp
src/group_manager.cpp src/group_manager.cpp
src/input_stream.cpp src/input_stream.cpp
src/ipv4_acceptor.cpp
src/ipv4_io_stream.cpp
src/local_actor.cpp src/local_actor.cpp
src/logging.cpp src/logging.cpp
src/mailbox_element.cpp src/mailbox_element.cpp
...@@ -203,11 +199,13 @@ set(LIBCPPA_SRC ...@@ -203,11 +199,13 @@ set(LIBCPPA_SRC
src/opt.cpp src/opt.cpp
src/output_stream.cpp src/output_stream.cpp
src/partial_function.cpp src/partial_function.cpp
src/peer.cpp
src/peer_acceptor.cpp
src/primitive_variant.cpp src/primitive_variant.cpp
src/ref_counted.cpp src/ref_counted.cpp
src/resumable.cpp
src/remote_actor_proxy.cpp src/remote_actor_proxy.cpp
src/response_promise.cpp src/response_promise.cpp
src/resumable.cpp
src/ripemd_160.cpp src/ripemd_160.cpp
src/scheduler.cpp src/scheduler.cpp
src/scoped_actor.cpp src/scoped_actor.cpp
...@@ -217,6 +215,8 @@ set(LIBCPPA_SRC ...@@ -217,6 +215,8 @@ set(LIBCPPA_SRC
src/stream.cpp src/stream.cpp
src/string_serialization.cpp src/string_serialization.cpp
src/sync_request_bouncer.cpp src/sync_request_bouncer.cpp
src/tcp_acceptor.cpp
src/tcp_io_stream.cpp
src/to_uniform_name.cpp src/to_uniform_name.cpp
src/type_lookup_table.cpp src/type_lookup_table.cpp
src/unicast_network.cpp src/unicast_network.cpp
......
...@@ -98,8 +98,6 @@ cppa/io/continuable.hpp ...@@ -98,8 +98,6 @@ cppa/io/continuable.hpp
cppa/io/default_message_queue.hpp cppa/io/default_message_queue.hpp
cppa/io/event.hpp cppa/io/event.hpp
cppa/io/input_stream.hpp cppa/io/input_stream.hpp
cppa/io/ipv4_acceptor.hpp
cppa/io/ipv4_io_stream.hpp
cppa/io/middleman.hpp cppa/io/middleman.hpp
cppa/io/middleman_event_handler.hpp cppa/io/middleman_event_handler.hpp
cppa/io/output_stream.hpp cppa/io/output_stream.hpp
...@@ -107,6 +105,8 @@ cppa/io/peer.hpp ...@@ -107,6 +105,8 @@ cppa/io/peer.hpp
cppa/io/peer_acceptor.hpp cppa/io/peer_acceptor.hpp
cppa/io/remote_actor_proxy.hpp cppa/io/remote_actor_proxy.hpp
cppa/io/stream.hpp cppa/io/stream.hpp
cppa/io/tcp_acceptor.hpp
cppa/io/tcp_io_stream.hpp
cppa/local_actor.hpp cppa/local_actor.hpp
cppa/logging.hpp cppa/logging.hpp
cppa/mailbox_based.hpp cppa/mailbox_based.hpp
...@@ -232,6 +232,7 @@ examples/remote_actors/distributed_calculator.cpp ...@@ -232,6 +232,7 @@ examples/remote_actors/distributed_calculator.cpp
examples/remote_actors/group_chat.cpp examples/remote_actors/group_chat.cpp
examples/remote_actors/group_server.cpp examples/remote_actors/group_server.cpp
examples/remote_actors/protobuf_broker.cpp examples/remote_actors/protobuf_broker.cpp
examples/remote_actors/simple_broker.cpp
examples/type_system/announce_1.cpp examples/type_system/announce_1.cpp
examples/type_system/announce_2.cpp examples/type_system/announce_2.cpp
examples/type_system/announce_3.cpp examples/type_system/announce_3.cpp
...@@ -285,8 +286,6 @@ src/get_root_uuid.cpp ...@@ -285,8 +286,6 @@ src/get_root_uuid.cpp
src/group.cpp src/group.cpp
src/group_manager.cpp src/group_manager.cpp
src/input_stream.cpp src/input_stream.cpp
src/ipv4_acceptor.cpp
src/ipv4_io_stream.cpp
src/local_actor.cpp src/local_actor.cpp
src/logging.cpp src/logging.cpp
src/mailbox_element.cpp src/mailbox_element.cpp
...@@ -325,6 +324,8 @@ src/singleton_manager.cpp ...@@ -325,6 +324,8 @@ src/singleton_manager.cpp
src/stream.cpp src/stream.cpp
src/string_serialization.cpp src/string_serialization.cpp
src/sync_request_bouncer.cpp src/sync_request_bouncer.cpp
src/tcp_acceptor.cpp
src/tcp_io_stream.cpp
src/to_uniform_name.cpp src/to_uniform_name.cpp
src/type_lookup_table.cpp src/type_lookup_table.cpp
src/unicast_network.cpp src/unicast_network.cpp
......
...@@ -72,8 +72,8 @@ ...@@ -72,8 +72,8 @@
#include "cppa/io/input_stream.hpp" #include "cppa/io/input_stream.hpp"
#include "cppa/io/output_stream.hpp" #include "cppa/io/output_stream.hpp"
#include "cppa/io/accept_handle.hpp" #include "cppa/io/accept_handle.hpp"
#include "cppa/io/ipv4_acceptor.hpp" #include "cppa/io/tcp_acceptor.hpp"
#include "cppa/io/ipv4_io_stream.hpp" #include "cppa/io/tcp_io_stream.hpp"
#include "cppa/io/connection_handle.hpp" #include "cppa/io/connection_handle.hpp"
#include "cppa/detail/memory.hpp" #include "cppa/detail/memory.hpp"
...@@ -522,7 +522,7 @@ struct typed_remote_actor_helper<util::type_list<Ts...>> { ...@@ -522,7 +522,7 @@ struct typed_remote_actor_helper<util::type_list<Ts...>> {
return res; return res;
} }
return_type operator()(const char* host, std::uint16_t port) { return_type operator()(const char* host, std::uint16_t port) {
auto ptr = io::ipv4_io_stream::connect_to(host, port); auto ptr = io::tcp_io_stream::connect_to(host, port);
return (*this)(io::stream_ptr_pair(ptr, ptr)); return (*this)(io::stream_ptr_pair(ptr, ptr));
} }
}; };
...@@ -596,7 +596,7 @@ void typed_publish(typed_actor<Rs...> whom, ...@@ -596,7 +596,7 @@ void typed_publish(typed_actor<Rs...> whom,
std::uint16_t port, const char* addr = nullptr) { std::uint16_t port, const char* addr = nullptr) {
if (!whom) return; if (!whom) return;
detail::publish_impl(detail::raw_access::get(whom), detail::publish_impl(detail::raw_access::get(whom),
io::ipv4_acceptor::create(port, addr)); io::tcp_acceptor::create(port, addr));
} }
/** /**
...@@ -667,7 +667,7 @@ template<spawn_options Os = no_spawn_options, ...@@ -667,7 +667,7 @@ template<spawn_options Os = no_spawn_options,
typename F = std::function<void (io::broker*)>, typename F = std::function<void (io::broker*)>,
typename... Ts> typename... Ts>
actor spawn_io(F fun, const std::string& host, uint16_t port, Ts&&... args) { actor spawn_io(F fun, const std::string& host, uint16_t port, Ts&&... args) {
auto ptr = io::ipv4_io_stream::connect_to(host.c_str(), port); auto ptr = io::tcp_io_stream::connect_to(host.c_str(), port);
return spawn_io(std::move(fun), ptr, ptr, std::forward<Ts>(args)...); return spawn_io(std::move(fun), ptr, ptr, std::forward<Ts>(args)...);
} }
...@@ -681,7 +681,7 @@ actor spawn_io_server(F fun, uint16_t port, Ts&&... args) { ...@@ -681,7 +681,7 @@ actor spawn_io_server(F fun, uint16_t port, Ts&&... args) {
"top-level spawns cannot have monitor or link flag"); "top-level spawns cannot have monitor or link flag");
using namespace std; using namespace std;
auto ptr = io::broker::from(move(fun), auto ptr = io::broker::from(move(fun),
io::ipv4_acceptor::create(port), io::tcp_acceptor::create(port),
forward<Ts>(args)...); forward<Ts>(args)...);
return {io::init_and_launch(move(ptr))}; return {io::init_and_launch(move(ptr))};
} }
......
...@@ -57,7 +57,7 @@ class acceptor { ...@@ -57,7 +57,7 @@ class acceptor {
/** /**
* @brief Returns the internal file descriptor. This descriptor is needed * @brief Returns the internal file descriptor. This descriptor is needed
* for socket multiplexing using select(). * for socket multiplexing using select()/poll()/epoll().
*/ */
virtual native_socket_type file_handle() const = 0; virtual native_socket_type file_handle() const = 0;
......
...@@ -39,8 +39,11 @@ ...@@ -39,8 +39,11 @@
#include "cppa/util/buffer.hpp" #include "cppa/util/buffer.hpp"
#include "cppa/io/stream.hpp"
#include "cppa/io/acceptor.hpp" #include "cppa/io/acceptor.hpp"
#include "cppa/io/input_stream.hpp" #include "cppa/io/input_stream.hpp"
#include "cppa/io/tcp_acceptor.hpp"
#include "cppa/io/tcp_io_stream.hpp"
#include "cppa/io/output_stream.hpp" #include "cppa/io/output_stream.hpp"
#include "cppa/io/accept_handle.hpp" #include "cppa/io/accept_handle.hpp"
#include "cppa/io/connection_handle.hpp" #include "cppa/io/connection_handle.hpp"
...@@ -149,6 +152,22 @@ class broker : public extend<local_actor>:: ...@@ -149,6 +152,22 @@ class broker : public extend<local_actor>::
return m_io.size(); return m_io.size();
} }
connection_handle add_connection(input_stream_ptr in, output_stream_ptr out);
inline connection_handle add_connection(stream_ptr sptr) {
return add_connection(sptr, sptr);
}
inline connection_handle add_tcp_connection(native_socket_type tcp_sockfd) {
return add_connection(tcp_io_stream::from_sockfd(tcp_sockfd));
}
accept_handle add_acceptor(acceptor_uptr ptr);
inline accept_handle add_tcp_acceptor(native_socket_type tcp_sockfd) {
return add_acceptor(tcp_acceptor::from_sockfd(tcp_sockfd));
}
protected: protected:
broker(input_stream_ptr in, output_stream_ptr out); broker(input_stream_ptr in, output_stream_ptr out);
...@@ -170,6 +189,9 @@ class broker : public extend<local_actor>:: ...@@ -170,6 +189,9 @@ class broker : public extend<local_actor>::
actor fork_impl(std::function<void (broker*)> fun, actor fork_impl(std::function<void (broker*)> fun,
connection_handle hdl); connection_handle hdl);
actor fork_impl(std::function<behavior (broker*)> fun,
connection_handle hdl);
static broker_ptr from_impl(std::function<void (broker*)> fun, static broker_ptr from_impl(std::function<void (broker*)> fun,
input_stream_ptr in, input_stream_ptr in,
output_stream_ptr out); output_stream_ptr out);
...@@ -188,10 +210,6 @@ class broker : public extend<local_actor>:: ...@@ -188,10 +210,6 @@ class broker : public extend<local_actor>::
void init_broker(); void init_broker();
connection_handle add_scribe(input_stream_ptr in, output_stream_ptr out);
accept_handle add_doorman(acceptor_uptr ptr);
std::map<accept_handle, doorman_pointer> m_accept; std::map<accept_handle, doorman_pointer> m_accept;
std::map<connection_handle, scribe_pointer> m_io; std::map<connection_handle, scribe_pointer> m_io;
......
...@@ -33,7 +33,7 @@ ...@@ -33,7 +33,7 @@
#include "cppa/actor.hpp" #include "cppa/actor.hpp"
#include "cppa/io/ipv4_acceptor.hpp" #include "cppa/io/tcp_acceptor.hpp"
#include "cppa/io/continuable.hpp" #include "cppa/io/continuable.hpp"
namespace cppa { namespace io { namespace cppa { namespace io {
......
...@@ -39,24 +39,40 @@ ...@@ -39,24 +39,40 @@
namespace cppa { namespace io { namespace cppa { namespace io {
class ipv4_acceptor : public acceptor { /**
* @brief An implementation of the {@link acceptor} interface for TCP sockets.
*/
class tcp_acceptor : public acceptor {
public: public:
/**
* @brief Creates an TCP acceptor and binds it to given @p port. Incoming
* connections are only accepted from the address @p addr.
* Per default, i.e., <tt>addr == nullptr</tt>, all incoming
* connections are accepted.
* @throws network_error if a socket operation fails
* @throws bind_failure if given port is already in use
*/
static std::unique_ptr<acceptor> create(std::uint16_t port, static std::unique_ptr<acceptor> create(std::uint16_t port,
const char* addr = nullptr); const char* addr = nullptr);
~ipv4_acceptor(); /**
* @brief Creates an TCP acceptor from the native socket handle @p fd.
*/
static std::unique_ptr<acceptor> from_sockfd(native_socket_type fd);
native_socket_type file_handle() const; ~tcp_acceptor();
stream_ptr_pair accept_connection(); native_socket_type file_handle() const override;
optional<stream_ptr_pair> try_accept_connection(); stream_ptr_pair accept_connection() override;
optional<stream_ptr_pair> try_accept_connection() override;
private: private:
ipv4_acceptor(native_socket_type fd, bool nonblocking); tcp_acceptor(native_socket_type fd, bool nonblocking);
native_socket_type m_fd; native_socket_type m_fd;
bool m_is_nonblocking; bool m_is_nonblocking;
......
...@@ -36,15 +36,25 @@ ...@@ -36,15 +36,25 @@
namespace cppa { namespace io { namespace cppa { namespace io {
class ipv4_io_stream : public stream { /**
* @brief An implementation of the {@link stream} interface for TCP sockets.
*/
class tcp_io_stream : public stream {
public: public:
~ipv4_io_stream(); ~tcp_io_stream();
/**
* @brief Establishes a TCP connection to given @p host at given @p port.
* @throws network_error if connection fails or read error occurs
*/
static stream_ptr connect_to(const char* host, std::uint16_t port); static stream_ptr connect_to(const char* host, std::uint16_t port);
static stream_ptr from_native_socket(native_socket_type fd); /**
* @brief Creates an TCP stream from the native socket handle @p fd.
*/
static stream_ptr from_sockfd(native_socket_type fd);
native_socket_type read_handle() const; native_socket_type read_handle() const;
...@@ -60,7 +70,7 @@ class ipv4_io_stream : public stream { ...@@ -60,7 +70,7 @@ class ipv4_io_stream : public stream {
private: private:
ipv4_io_stream(native_socket_type fd); tcp_io_stream(native_socket_type fd);
native_socket_type m_fd; native_socket_type m_fd;
......
...@@ -23,6 +23,7 @@ add(typed_calculator message_passing) ...@@ -23,6 +23,7 @@ add(typed_calculator message_passing)
add(distributed_calculator remote_actors) add(distributed_calculator remote_actors)
add(group_server remote_actors) add(group_server remote_actors)
add(group_chat remote_actors) add(group_chat remote_actors)
add(simple_broker remote_actors)
if (NOT "${CPPA_NO_PROTOBUF_EXAMPLES}" STREQUAL "yes") if (NOT "${CPPA_NO_PROTOBUF_EXAMPLES}" STREQUAL "yes")
find_package(Protobuf) find_package(Protobuf)
......
...@@ -42,8 +42,8 @@ ...@@ -42,8 +42,8 @@
#include "cppa/io/broker.hpp" #include "cppa/io/broker.hpp"
#include "cppa/io/middleman.hpp" #include "cppa/io/middleman.hpp"
#include "cppa/io/ipv4_acceptor.hpp" #include "cppa/io/tcp_acceptor.hpp"
#include "cppa/io/ipv4_io_stream.hpp" #include "cppa/io/tcp_io_stream.hpp"
CPPA_PUSH_WARNINGS CPPA_PUSH_WARNINGS
#include "pingpong.pb.h" #include "pingpong.pb.h"
......
/******************************************************************************\
* This example program showcases how to manually manage socket IO using *
* a broker. Server and client exchange integers in a 'ping-pong protocol'. *
* *
* Minimal setup: *
* - ./build/bin/broker -s 4242 *
* - ./build/bin/broker -c localhost 4242 *
\******************************************************************************/
#include <vector>
#include <string>
#include <limits>
#include <memory>
#include <cstdint>
#include <cassert>
#include <iostream>
#include <arpa/inet.h> // htonl
#include "cppa/cppa.hpp"
using namespace std;
using namespace cppa;
using namespace cppa::io;
// utility function to print an exit message with custom name
void print_on_exit(const actor& ptr, const std::string& name) {
ptr->attach_functor([=](std::uint32_t reason) {
aout(ptr) << name << " exited with reason " << reason << endl;
});
}
behavior ping(event_based_actor* self, size_t num_pings) {
auto count = make_shared<size_t>(0);
return {
on(atom("kickoff"), arg_match) >> [=](const actor& pong) {
self->send(pong, atom("ping"), int32_t(1));
self->become (
on(atom("pong"), arg_match) >> [=](int32_t value) -> any_tuple {
if (++*count >= num_pings) self->quit();
return make_any_tuple(atom("ping"), value + 1);
}
);
}
};
}
behavior pong() {
return {
on(atom("ping"), arg_match) >> [](int32_t value) {
return make_any_tuple(atom("pong"), value);
}
};
}
// utility function for sending an integer type
template<typename T>
void write_int(broker* self, connection_handle hdl, T value) {
auto cpy = static_cast<T>(htonl(value));
self->write(hdl, sizeof(T), &cpy);
}
// utility function for reading an ingeger from incoming data
template<typename T>
T read_int(const void* data) {
T value;
memcpy(&value, data, sizeof(T));
return static_cast<T>(ntohl(value));
}
// implemenation of our broker
behavior broker_impl(broker* self, connection_handle hdl, const actor& buddy) {
// we assume io_fsm manages a broker with exactly one connection,
// i.e., the connection ponted to by `hdl`
assert(self->num_connections() == 1);
// monitor buddy to quit broker if buddy is done
self->monitor(buddy);
// setup: we are exchanging only messages consisting of an atom
// (as uint64_t) and an integer value (int32_t)
self->receive_policy(hdl, broker::exactly,
sizeof(uint64_t) + sizeof(int32_t));
// our message handlers
return {
[=](const connection_closed_msg& msg) {
// brokers can multiplex any number of connections, however
// this example assumes io_fsm to manage a broker with
// exactly one connection
assert(msg.handle == hdl);
aout(self) << "connection closed" << endl;
// force buddy to quit
self->send_exit(buddy, exit_reason::remote_link_unreachable);
self->quit(exit_reason::remote_link_unreachable);
},
[=](atom_value av, int32_t i) {
assert(av == atom("ping") || av == atom("pong"));
aout(self) << "send {" << to_string(av) << ", " << i << "}"
<< endl;
// cast atom to its underlying type, i.e., uint64_t
write_int(self, hdl, static_cast<uint64_t>(av));
write_int(self, hdl, i);
},
[=](const down_msg& dm) {
if (dm.source == buddy) {
aout(self) << "our buddy is down" << endl;
// quit for same reason
self->quit(dm.reason);
}
},
[=](const new_data_msg& msg) {
// read the atom value as uint64_t from buffer
auto atm_val = read_int<uint64_t>(msg.buf.data());
// cast to original type
auto atm = static_cast<atom_value>(atm_val);
// read integer value from buffer, jumping to the correct
// position via offset_data(...)
auto ival = read_int<int32_t>(msg.buf.offset_data(sizeof(uint64_t)));
// show some output
aout(self) << "received {" << to_string(atm) << ", " << ival << "}"
<< endl;
// send composed message to our buddy
self->send(buddy, atm, ival);
},
others() >> [=] {
cout << "unexpected: " << to_string(self->last_dequeued()) << endl;
}
};
}
behavior server(broker* self, const actor& buddy) {
aout(self) << "server is running" << endl;
return {
[=](const new_connection_msg& msg) {
aout(self) << "server accepted new connection" << endl;
// by forking into a new broker, we are no longer
// responsible for the connection
auto impl = self->fork(broker_impl, msg.handle, buddy);
print_on_exit(impl, "broker_impl");
aout(self) << "quit server (only accept 1 connection)" << endl;
self->quit();
},
others() >> [=] {
cout << "unexpected: " << to_string(self->last_dequeued()) << endl;
}
};
}
optional<uint16_t> as_u16(const std::string& str) {
return static_cast<uint16_t>(stoul(str));
}
int main(int argc, char** argv) {
match(std::vector<string>{argv + 1, argv + argc}) (
on("-s", as_u16) >> [&](uint16_t port) {
cout << "run in server mode" << endl;
auto pong_actor = spawn(pong);
auto sever_actor = spawn_io_server(server, port, pong_actor);
print_on_exit(sever_actor, "server");
print_on_exit(pong_actor, "pong");
},
on("-c", val<string>, as_u16) >> [&](const string& host, uint16_t port) {
auto ping_actor = spawn(ping, 20);
auto io_actor = spawn_io(broker_impl, host, port, ping_actor);
print_on_exit(io_actor, "protobuf_io");
print_on_exit(ping_actor, "ping");
send_as(io_actor, ping_actor, atom("kickoff"), io_actor);
},
others() >> [] {
cerr << "use with eihter '-s PORT' as server or '-c HOST PORT' as client"
<< endl;
}
);
await_all_actors_done();
shutdown();
}
...@@ -281,8 +281,8 @@ class broker::doorman : public broker::servant { ...@@ -281,8 +281,8 @@ class broker::doorman : public broker::servant {
if (opt) { if (opt) {
using namespace std; using namespace std;
auto& p = *opt; auto& p = *opt;
get_ref<0>(m_accept_msg).handle = m_broker->add_scribe(move(p.first), get_ref<0>(m_accept_msg).handle = m_broker->add_connection(move(p.first),
move(p.second)); move(p.second));
m_broker->invoke_message({invalid_actor_addr, nullptr}, m_accept_msg); m_broker->invoke_message({invalid_actor_addr, nullptr}, m_accept_msg);
} }
else return continue_reading_result::continue_later; else return continue_reading_result::continue_later;
...@@ -408,7 +408,7 @@ void broker::init_broker() { ...@@ -408,7 +408,7 @@ void broker::init_broker() {
broker::broker(input_stream_ptr in, output_stream_ptr out) { broker::broker(input_stream_ptr in, output_stream_ptr out) {
using namespace std; using namespace std;
init_broker(); init_broker();
add_scribe(move(in), move(out)); add_connection(move(in), move(out));
} }
broker::broker(scribe_pointer ptr) { broker::broker(scribe_pointer ptr) {
...@@ -421,7 +421,7 @@ broker::broker(scribe_pointer ptr) { ...@@ -421,7 +421,7 @@ broker::broker(scribe_pointer ptr) {
broker::broker(acceptor_uptr ptr) { broker::broker(acceptor_uptr ptr) {
using namespace std; using namespace std;
init_broker(); init_broker();
add_doorman(move(ptr)); add_acceptor(move(ptr));
} }
void broker::cleanup(std::uint32_t reason) { void broker::cleanup(std::uint32_t reason) {
...@@ -501,21 +501,21 @@ void broker::erase_acceptor(int id) { ...@@ -501,21 +501,21 @@ void broker::erase_acceptor(int id) {
m_accept.erase(accept_handle::from_int(id)); m_accept.erase(accept_handle::from_int(id));
} }
connection_handle broker::add_scribe(input_stream_ptr in, output_stream_ptr out) { connection_handle broker::add_connection(input_stream_ptr in, output_stream_ptr out) {
using namespace std; using namespace std;
auto id = connection_handle::from_int(in->read_handle()); auto id = connection_handle::from_int(in->read_handle());
m_io.insert(make_pair(id, create_unique<scribe>(this, move(in), move(out)))); m_io.insert(make_pair(id, create_unique<scribe>(this, move(in), move(out))));
return id; return id;
} }
accept_handle broker::add_doorman(acceptor_uptr ptr) { accept_handle broker::add_acceptor(acceptor_uptr ptr) {
using namespace std; using namespace std;
auto id = accept_handle::from_int(ptr->file_handle()); auto id = accept_handle::from_int(ptr->file_handle());
m_accept.insert(make_pair(id, create_unique<doorman>(this, move(ptr)))); m_accept.insert(make_pair(id, create_unique<doorman>(this, move(ptr))));
return id; return id;
} }
actor broker::fork_impl(std::function<void (broker*)> fun, actor broker::fork_impl(std::function<behavior (broker*)> fun,
connection_handle hdl) { connection_handle hdl) {
CPPA_LOG_TRACE(CPPA_MARG(hdl, id)); CPPA_LOG_TRACE(CPPA_MARG(hdl, id));
auto i = m_io.find(hdl); auto i = m_io.find(hdl);
...@@ -524,14 +524,21 @@ actor broker::fork_impl(std::function<void (broker*)> fun, ...@@ -524,14 +524,21 @@ actor broker::fork_impl(std::function<void (broker*)> fun,
throw std::invalid_argument("invalid handle"); throw std::invalid_argument("invalid handle");
} }
scribe* sptr = i->second.get(); // non-owning pointer scribe* sptr = i->second.get(); // non-owning pointer
auto f = [=](broker* ptr) -> behavior { fun(ptr); return behavior{}; }; auto result = make_counted<default_broker>(fun, move(i->second));
auto result = make_counted<default_broker>(f, move(i->second));
init_and_launch(result); init_and_launch(result);
sptr->set_broker(result); // set new broker sptr->set_broker(result); // set new broker
m_io.erase(i); m_io.erase(i);
return {result}; return {result};
} }
actor broker::fork_impl(std::function<void (broker*)> fun,
connection_handle hdl) {
return fork_impl([=](broker* self) -> behavior {
fun(self);
return behavior{};
}, hdl);
}
void broker::receive_policy(const connection_handle& hdl, void broker::receive_policy(const connection_handle& hdl,
broker::policy_flag policy, broker::policy_flag policy,
size_t buffer_size) { size_t buffer_size) {
......
...@@ -37,8 +37,8 @@ ...@@ -37,8 +37,8 @@
#include "cppa/exception.hpp" #include "cppa/exception.hpp"
#include "cppa/io/stream.hpp" #include "cppa/io/stream.hpp"
#include "cppa/io/ipv4_acceptor.hpp" #include "cppa/io/tcp_acceptor.hpp"
#include "cppa/io/ipv4_io_stream.hpp" #include "cppa/io/tcp_io_stream.hpp"
#include "cppa/detail/fd_util.hpp" #include "cppa/detail/fd_util.hpp"
...@@ -95,7 +95,7 @@ bool accept_impl(stream_ptr_pair& result, ...@@ -95,7 +95,7 @@ bool accept_impl(stream_ptr_pair& result,
} }
throw_io_failure("accept failed"); throw_io_failure("accept failed");
} }
stream_ptr ptr(ipv4_io_stream::from_native_socket(sfd)); stream_ptr ptr(tcp_io_stream::from_sockfd(sfd));
result.first = ptr; result.first = ptr;
result.second = ptr; result.second = ptr;
return true; return true;
...@@ -103,12 +103,12 @@ bool accept_impl(stream_ptr_pair& result, ...@@ -103,12 +103,12 @@ bool accept_impl(stream_ptr_pair& result,
} // namespace <anonymous> } // namespace <anonymous>
ipv4_acceptor::ipv4_acceptor(native_socket_type fd, bool nonblocking) tcp_acceptor::tcp_acceptor(native_socket_type fd, bool nonblocking)
: m_fd(fd), m_is_nonblocking(nonblocking) { } : m_fd(fd), m_is_nonblocking(nonblocking) { }
std::unique_ptr<acceptor> ipv4_acceptor::create(std::uint16_t port, std::unique_ptr<acceptor> tcp_acceptor::create(std::uint16_t port,
const char* addr) { const char* addr) {
CPPA_LOGM_TRACE("ipv4_acceptor", CPPA_ARG(port) << ", addr = " CPPA_LOGM_TRACE("tcp_acceptor", CPPA_ARG(port) << ", addr = "
<< (addr ? addr : "nullptr")); << (addr ? addr : "nullptr"));
# ifdef CPPA_WINDOWS # ifdef CPPA_WINDOWS
// ensure that TCP has been initialized via WSAStartup // ensure that TCP has been initialized via WSAStartup
...@@ -141,23 +141,32 @@ std::unique_ptr<acceptor> ipv4_acceptor::create(std::uint16_t port, ...@@ -141,23 +141,32 @@ std::unique_ptr<acceptor> ipv4_acceptor::create(std::uint16_t port,
if (listen(sockfd, 10) != 0) { if (listen(sockfd, 10) != 0) {
throw network_error("listen() failed"); throw network_error("listen() failed");
} }
// ok, no exceptions so far
sguard.release();
CPPA_LOGM_DEBUG("tcp_acceptor", "sockfd = " << sockfd);
return from_sockfd(sockfd);
}
std::unique_ptr<acceptor> tcp_acceptor::from_sockfd(native_socket_type fd) {
// sguard closes the socket in case of exception
socket_guard sguard(fd);
// default mode is nonblocking // default mode is nonblocking
nonblocking(sockfd, true); nonblocking(fd, true);
// ok, no exceptions // ok, no exceptions
sguard.release(); sguard.release();
CPPA_LOGM_DEBUG("ipv4_acceptor", "sockfd = " << sockfd); return std::unique_ptr<tcp_acceptor>(new tcp_acceptor(fd, true));
return std::unique_ptr<ipv4_acceptor>(new ipv4_acceptor(sockfd, true));
} }
ipv4_acceptor::~ipv4_acceptor() {
tcp_acceptor::~tcp_acceptor() {
closesocket(m_fd); closesocket(m_fd);
} }
native_socket_type ipv4_acceptor::file_handle() const { native_socket_type tcp_acceptor::file_handle() const {
return m_fd; return m_fd;
} }
stream_ptr_pair ipv4_acceptor::accept_connection() { stream_ptr_pair tcp_acceptor::accept_connection() {
if (m_is_nonblocking) { if (m_is_nonblocking) {
nonblocking(m_fd, false); nonblocking(m_fd, false);
m_is_nonblocking = false; m_is_nonblocking = false;
...@@ -167,7 +176,7 @@ stream_ptr_pair ipv4_acceptor::accept_connection() { ...@@ -167,7 +176,7 @@ stream_ptr_pair ipv4_acceptor::accept_connection() {
return result; return result;
} }
optional<stream_ptr_pair> ipv4_acceptor::try_accept_connection() { optional<stream_ptr_pair> tcp_acceptor::try_accept_connection() {
if (!m_is_nonblocking) { if (!m_is_nonblocking) {
nonblocking(m_fd, true); nonblocking(m_fd, true);
m_is_nonblocking = true; m_is_nonblocking = true;
......
...@@ -37,7 +37,7 @@ ...@@ -37,7 +37,7 @@
#include "cppa/logging.hpp" #include "cppa/logging.hpp"
#include "cppa/exception.hpp" #include "cppa/exception.hpp"
#include "cppa/detail/fd_util.hpp" #include "cppa/detail/fd_util.hpp"
#include "cppa/io/ipv4_io_stream.hpp" #include "cppa/io/tcp_io_stream.hpp"
#ifdef CPPA_WINDOWS #ifdef CPPA_WINDOWS
# include <ws2tcpip.h> # include <ws2tcpip.h>
...@@ -55,21 +55,21 @@ namespace cppa { namespace io { ...@@ -55,21 +55,21 @@ namespace cppa { namespace io {
using namespace ::cppa::detail::fd_util; using namespace ::cppa::detail::fd_util;
ipv4_io_stream::ipv4_io_stream(native_socket_type fd) : m_fd(fd) { } tcp_io_stream::tcp_io_stream(native_socket_type fd) : m_fd(fd) { }
ipv4_io_stream::~ipv4_io_stream() { tcp_io_stream::~tcp_io_stream() {
closesocket(m_fd); closesocket(m_fd);
} }
native_socket_type ipv4_io_stream::read_handle() const { native_socket_type tcp_io_stream::read_handle() const {
return m_fd; return m_fd;
} }
native_socket_type ipv4_io_stream::write_handle() const { native_socket_type tcp_io_stream::write_handle() const {
return m_fd; return m_fd;
} }
void ipv4_io_stream::read(void* vbuf, size_t len) { void tcp_io_stream::read(void* vbuf, size_t len) {
auto buf = reinterpret_cast<char*>(vbuf); auto buf = reinterpret_cast<char*>(vbuf);
size_t rd = 0; size_t rd = 0;
while (rd < len) { while (rd < len) {
...@@ -89,13 +89,13 @@ void ipv4_io_stream::read(void* vbuf, size_t len) { ...@@ -89,13 +89,13 @@ void ipv4_io_stream::read(void* vbuf, size_t len) {
} }
} }
size_t ipv4_io_stream::read_some(void* buf, size_t len) { size_t tcp_io_stream::read_some(void* buf, size_t len) {
auto recv_result = ::recv(m_fd, reinterpret_cast<char*>(buf), len, 0); auto recv_result = ::recv(m_fd, reinterpret_cast<char*>(buf), len, 0);
handle_read_result(recv_result, true); handle_read_result(recv_result, true);
return (recv_result > 0) ? static_cast<size_t>(recv_result) : 0; return (recv_result > 0) ? static_cast<size_t>(recv_result) : 0;
} }
void ipv4_io_stream::write(const void* vbuf, size_t len) { void tcp_io_stream::write(const void* vbuf, size_t len) {
auto buf = reinterpret_cast<const char*>(vbuf); auto buf = reinterpret_cast<const char*>(vbuf);
size_t written = 0; size_t written = 0;
while (written < len) { while (written < len) {
...@@ -116,20 +116,20 @@ void ipv4_io_stream::write(const void* vbuf, size_t len) { ...@@ -116,20 +116,20 @@ void ipv4_io_stream::write(const void* vbuf, size_t len) {
} }
} }
size_t ipv4_io_stream::write_some(const void* buf, size_t len) { size_t tcp_io_stream::write_some(const void* buf, size_t len) {
CPPA_LOG_TRACE(CPPA_ARG(buf) << ", " << CPPA_ARG(len)); CPPA_LOG_TRACE(CPPA_ARG(buf) << ", " << CPPA_ARG(len));
auto send_result = ::send(m_fd, reinterpret_cast<const char*>(buf), len, 0); auto send_result = ::send(m_fd, reinterpret_cast<const char*>(buf), len, 0);
handle_write_result(send_result, true); handle_write_result(send_result, true);
return static_cast<size_t>(send_result); return static_cast<size_t>(send_result);
} }
io::stream_ptr ipv4_io_stream::from_native_socket(native_socket_type fd) { io::stream_ptr tcp_io_stream::from_sockfd(native_socket_type fd) {
tcp_nodelay(fd, true); tcp_nodelay(fd, true);
nonblocking(fd, true); nonblocking(fd, true);
return new ipv4_io_stream(fd); return new tcp_io_stream(fd);
} }
io::stream_ptr ipv4_io_stream::connect_to(const char* host, io::stream_ptr tcp_io_stream::connect_to(const char* host,
std::uint16_t port) { std::uint16_t port) {
CPPA_LOGF_TRACE(CPPA_ARG(host) << ", " << CPPA_ARG(port)); CPPA_LOGF_TRACE(CPPA_ARG(host) << ", " << CPPA_ARG(port));
CPPA_LOGF_INFO("try to connect to " << host << " on port " << port); CPPA_LOGF_INFO("try to connect to " << host << " on port " << port);
...@@ -162,9 +162,7 @@ io::stream_ptr ipv4_io_stream::connect_to(const char* host, ...@@ -162,9 +162,7 @@ io::stream_ptr ipv4_io_stream::connect_to(const char* host,
throw network_error("could not connect to host"); throw network_error("could not connect to host");
} }
CPPA_LOGF_DEBUG("enable nodelay + nonblocking for socket"); CPPA_LOGF_DEBUG("enable nodelay + nonblocking for socket");
tcp_nodelay(fd, true); return from_sockfd(fd);
nonblocking(fd, true);
return new ipv4_io_stream(fd);
} }
} } // namespace cppa::detail } } // namespace cppa::detail
...@@ -58,8 +58,8 @@ ...@@ -58,8 +58,8 @@
#include "cppa/io/acceptor.hpp" #include "cppa/io/acceptor.hpp"
#include "cppa/io/middleman.hpp" #include "cppa/io/middleman.hpp"
#include "cppa/io/peer_acceptor.hpp" #include "cppa/io/peer_acceptor.hpp"
#include "cppa/io/ipv4_acceptor.hpp" #include "cppa/io/tcp_acceptor.hpp"
#include "cppa/io/ipv4_io_stream.hpp" #include "cppa/io/tcp_io_stream.hpp"
#include "cppa/io/remote_actor_proxy.hpp" #include "cppa/io/remote_actor_proxy.hpp"
namespace { namespace {
...@@ -78,7 +78,7 @@ using namespace io; ...@@ -78,7 +78,7 @@ using namespace io;
void publish(actor whom, std::uint16_t port, const char* addr) { void publish(actor whom, std::uint16_t port, const char* addr) {
if (!whom) return; if (!whom) return;
publish(std::move(whom), io::ipv4_acceptor::create(port, addr)); publish(std::move(whom), io::tcp_acceptor::create(port, addr));
} }
void publish(actor whom, std::unique_ptr<io::acceptor> acceptor) { void publish(actor whom, std::unique_ptr<io::acceptor> acceptor) {
...@@ -91,7 +91,7 @@ actor remote_actor(io::stream_ptr_pair conn) { ...@@ -91,7 +91,7 @@ actor remote_actor(io::stream_ptr_pair conn) {
} }
actor remote_actor(const char* host, std::uint16_t port) { actor remote_actor(const char* host, std::uint16_t port) {
auto ptr = ipv4_io_stream::connect_to(host, port); auto ptr = tcp_io_stream::connect_to(host, port);
return remote_actor(stream_ptr_pair(ptr, ptr)); return remote_actor(stream_ptr_pair(ptr, ptr));
} }
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment