Commit d408143f authored by Joseph Noir's avatar Joseph Noir

Allow accept polices to spawn servers for mpxing

parent 589c5a2d
...@@ -8,6 +8,7 @@ file(GLOB_RECURSE LIBCAF_IO_HDRS "caf/*.hpp") ...@@ -8,6 +8,7 @@ file(GLOB_RECURSE LIBCAF_IO_HDRS "caf/*.hpp")
# list cpp files excluding platform-dependent files # list cpp files excluding platform-dependent files
set(LIBCAF_IO_SRCS set(LIBCAF_IO_SRCS
src/abstract_broker.cpp src/abstract_broker.cpp
src/acceptor_base.cpp
src/acceptor_manager.cpp src/acceptor_manager.cpp
src/acceptor_manager.cpp src/acceptor_manager.cpp
src/basp_broker.cpp src/basp_broker.cpp
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include "caf/io/network/newb_base.hpp"
#include "caf/policy/transport.hpp"
namespace caf {
namespace io {
namespace network {
struct acceptor_base : public newb_base {
acceptor_base(actor_config& cfg, default_multiplexer& dm,
native_socket sockfd);
virtual expected<actor>
create_newb(network::native_socket sockfd, policy::transport_ptr pol,
bool add_children_to_loop) = 0;
};
} // namespace network
} // namespace io
} // namespace caf
...@@ -31,6 +31,7 @@ ...@@ -31,6 +31,7 @@
#include "caf/detail/socket_guard.hpp" #include "caf/detail/socket_guard.hpp"
#include "caf/io/broker.hpp" #include "caf/io/broker.hpp"
#include "caf/io/middleman.hpp" #include "caf/io/middleman.hpp"
#include "caf/io/network/acceptor_base.hpp"
#include "caf/io/network/default_multiplexer.hpp" #include "caf/io/network/default_multiplexer.hpp"
#include "caf/io/network/event_handler.hpp" #include "caf/io/network/event_handler.hpp"
#include "caf/io/network/interfaces.hpp" #include "caf/io/network/interfaces.hpp"
...@@ -421,7 +422,7 @@ spawn_client(actor_system& sys, F fun, policy::transport_ptr transport, ...@@ -421,7 +422,7 @@ spawn_client(actor_system& sys, F fun, policy::transport_ptr transport,
// -- new broker acceptor ------------------------------------------------------ // -- new broker acceptor ------------------------------------------------------
template <class Protocol, class Fun, class... Ts> template <class Protocol, class Fun, class... Ts>
struct newb_acceptor : network::newb_base { struct newb_acceptor : network::acceptor_base {
using newb_type = typename std::remove_pointer<first_argument_type<Fun>>::type; using newb_type = typename std::remove_pointer<first_argument_type<Fun>>::type;
using message_type = typename newb_type::message_type; using message_type = typename newb_type::message_type;
...@@ -431,7 +432,7 @@ struct newb_acceptor : network::newb_base { ...@@ -431,7 +432,7 @@ struct newb_acceptor : network::newb_base {
network::native_socket sockfd, Fun f, network::native_socket sockfd, Fun f,
policy::accept_ptr<message_type> pol, policy::accept_ptr<message_type> pol,
std::tuple<Ts...> args) std::tuple<Ts...> args)
: newb_base(cfg, dm, sockfd), : acceptor_base(cfg, dm, sockfd),
accept_pol(std::move(pol)), accept_pol(std::move(pol)),
fun_(std::move(f)), fun_(std::move(f)),
reading_(false), reading_(false),
...@@ -562,7 +563,7 @@ struct newb_acceptor : network::newb_base { ...@@ -562,7 +563,7 @@ struct newb_acceptor : network::newb_base {
CAF_LOG_ERROR("failed to create socket for new endpoint"); CAF_LOG_ERROR("failed to create socket for new endpoint");
return; return;
} }
auto en = create_newb(sock, std::move(transport)); auto en = create_newb(sock, std::move(transport), true);
if (!en) { if (!en) {
io_error(network::operation::read, std::move(en.error())); io_error(network::operation::read, std::move(en.error()));
return; return;
...@@ -578,15 +579,16 @@ struct newb_acceptor : network::newb_base { ...@@ -578,15 +579,16 @@ struct newb_acceptor : network::newb_base {
accept_pol->write_event(this); accept_pol->write_event(this);
} }
virtual expected<actor> create_newb(network::native_socket sockfd, expected<actor> create_newb(network::native_socket sockfd,
policy::transport_ptr pol) { policy::transport_ptr pol,
bool add_children_to_loop) override {
CAF_LOG_TRACE(CAF_ARG(sockfd)); CAF_LOG_TRACE(CAF_ARG(sockfd));
auto n = detail::apply_args_prefixed( auto n = detail::apply_args_prefixed(
io::spawn_newb<Protocol, no_spawn_options, Fun, Ts...>, io::spawn_newb<Protocol, no_spawn_options, Fun, Ts...>,
detail::get_indices(args_), detail::get_indices(args_),
args_, this->backend().system(), args_, this->backend().system(),
fun_, std::move(pol), sockfd, fun_, std::move(pol), sockfd,
accept_pol->add_children_to_loop() add_children_to_loop
); );
link_to(n); link_to(n);
children_.push_back(n); children_.push_back(n);
......
...@@ -21,17 +21,10 @@ ...@@ -21,17 +21,10 @@
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/expected.hpp" #include "caf/expected.hpp"
#include "caf/io/network/native_socket.hpp" #include "caf/io/network/native_socket.hpp"
#include "caf/io/network/newb_base.hpp" #include "caf/io/network/acceptor_base.hpp"
#include "caf/policy/transport.hpp" #include "caf/policy/transport.hpp"
namespace caf { namespace caf {
namespace io {
template<class Message>
struct newb;
} // namespace io
namespace policy { namespace policy {
template <class Message> template <class Message>
...@@ -45,34 +38,30 @@ struct accept { ...@@ -45,34 +38,30 @@ struct accept {
// nop // nop
} }
virtual bool add_children_to_loop() {
return true;
}
virtual expected<io::network::native_socket> virtual expected<io::network::native_socket>
create_socket(uint16_t port, const char* host, bool reuse = false) = 0; create_socket(uint16_t port, const char* host, bool reuse = false) = 0;
virtual std::pair<io::network::native_socket, transport_ptr> virtual std::pair<io::network::native_socket, transport_ptr>
accept_event(io::network::newb_base*) { accept_event(io::network::acceptor_base*) {
return {0, nullptr}; return {0, nullptr};
} }
/// If `requires_raw_data` is set to true, the acceptor will only call /// If `requires_raw_data` is set to true, the acceptor will only call
/// this function for new read event and let the policy handle everything /// this function for new read event and let the policy handle everything
/// else. /// else.
virtual void read_event(io::network::newb_base*) { virtual void read_event(io::network::acceptor_base*) {
// nop // nop
} }
virtual error write_event(io::network::newb_base*) { virtual error write_event(io::network::acceptor_base*) {
return none; return none;
} }
virtual void init(io::network::newb_base*, io::newb<Message>&) { virtual void init(io::network::acceptor_base*, io::newb<Message>&) {
// nop // nop
} }
virtual void shutdown(io::network::newb_base*, virtual void shutdown(io::network::acceptor_base*,
io::network::native_socket) { io::network::native_socket) {
// nop // nop
} }
......
...@@ -18,6 +18,7 @@ ...@@ -18,6 +18,7 @@
#pragma once #pragma once
#include "caf/io/network/acceptor_base.hpp"
#include "caf/io/network/default_multiplexer.hpp" #include "caf/io/network/default_multiplexer.hpp"
#include "caf/io/network/native_socket.hpp" #include "caf/io/network/native_socket.hpp"
#include "caf/io/network/newb_base.hpp" #include "caf/io/network/newb_base.hpp"
...@@ -71,7 +72,7 @@ struct accept_tcp : public accept<Message> { ...@@ -71,7 +72,7 @@ struct accept_tcp : public accept<Message> {
} }
std::pair<io::network::native_socket, transport_ptr> std::pair<io::network::native_socket, transport_ptr>
accept_event(io::network::newb_base* parent) override { accept_event(io::network::acceptor_base* parent) override {
auto esock = io::network::accept_tcp_connection(get_newb_socket(parent)); auto esock = io::network::accept_tcp_connection(get_newb_socket(parent));
if (!esock) { if (!esock) {
return {io::network::invalid_native_socket, nullptr}; return {io::network::invalid_native_socket, nullptr};
...@@ -80,11 +81,11 @@ struct accept_tcp : public accept<Message> { ...@@ -80,11 +81,11 @@ struct accept_tcp : public accept<Message> {
return {*esock, std::move(ptr)}; return {*esock, std::move(ptr)};
} }
void init(io::network::newb_base*, io::newb<Message>& spawned) override { void init(io::network::acceptor_base*, io::newb<Message>& spawned) override {
spawned.start(); spawned.start();
} }
void shutdown(io::network::newb_base*, void shutdown(io::network::acceptor_base*,
io::network::native_socket sockfd) override { io::network::native_socket sockfd) override {
io::network::shutdown_both(sockfd); io::network::shutdown_both(sockfd);
} }
......
...@@ -18,6 +18,7 @@ ...@@ -18,6 +18,7 @@
#pragma once #pragma once
#include "caf/io/network/acceptor_base.hpp"
#include "caf/io/network/default_multiplexer.hpp" #include "caf/io/network/default_multiplexer.hpp"
#include "caf/io/network/ip_endpoint.hpp" #include "caf/io/network/ip_endpoint.hpp"
#include "caf/io/network/native_socket.hpp" #include "caf/io/network/native_socket.hpp"
...@@ -85,7 +86,7 @@ struct accept_udp : public accept<Message> { ...@@ -85,7 +86,7 @@ struct accept_udp : public accept<Message> {
} }
std::pair<io::network::native_socket, transport_ptr> std::pair<io::network::native_socket, transport_ptr>
accept_event(io::network::newb_base*) override { accept_event(io::network::acceptor_base*) override {
std::cout << "Accepting msg from UDP endpoint" << std::endl; std::cout << "Accepting msg from UDP endpoint" << std::endl;
auto res = io::network::new_local_udp_endpoint_impl(0, nullptr, true); auto res = io::network::new_local_udp_endpoint_impl(0, nullptr, true);
if (!res) { if (!res) {
...@@ -97,14 +98,14 @@ struct accept_udp : public accept<Message> { ...@@ -97,14 +98,14 @@ struct accept_udp : public accept<Message> {
return {sock, std::move(ptr)}; return {sock, std::move(ptr)};
} }
void init(io::network::newb_base* parent, void init(io::network::acceptor_base* parent,
io::newb<Message>& spawned) override { io::newb<Message>& spawned) override {
spawned.trans->prepare_next_read(parent); spawned.trans->prepare_next_read(parent);
spawned.trans->read_some(parent, *spawned.proto.get()); spawned.trans->read_some(parent, *spawned.proto.get());
spawned.start(); spawned.start();
} }
void shutdown(io::network::newb_base* parent, void shutdown(io::network::acceptor_base* parent,
io::network::native_socket) override { io::network::native_socket) override {
parent->passivate(); parent->passivate();
} }
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2018 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/io/network/acceptor_base.hpp"
namespace caf {
namespace io {
namespace network {
acceptor_base::acceptor_base(actor_config& cfg, default_multiplexer& dm,
native_socket sockfd)
: newb_base(cfg, dm, sockfd) {
// nop
}
} // namespace network
} // namespace io
} // namespace caf
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