Commit fbcedf80 authored by Joseph Noir's avatar Joseph Noir

Update init fun after accept to allow more config

parent 441360e3
......@@ -284,6 +284,8 @@ new_local_udp_endpoint_impl(uint16_t port, const char* addr,
bool reuse_addr = false,
optional<protocol::network> preferred = none);
expected<native_socket> accept_tcp_connection(native_socket sockfd);
} // namespace network
} // namespace io
} // namespace caf
......@@ -533,7 +533,7 @@ struct newb_acceptor : public newb_base, public caf::ref_counted {
// -- constructors and destructors -------------------------------------------
newb_acceptor(network::default_multiplexer& dm, network::native_socket sockfd,
Fun f, policy::accept_ptr pol, Ts&&... xs)
Fun f, policy::accept_ptr<message_type> pol, Ts&&... xs)
: newb_base(dm, sockfd),
accept_pol(std::move(pol)),
fun_(std::move(f)),
......@@ -541,6 +541,9 @@ struct newb_acceptor : public newb_base, public caf::ref_counted {
writing_(false),
args_(std::forward<Ts>(xs)...) {
// nop
if (sockfd == io::network::invalid_native_socket) {
std::cerr << "Creating newb with invalid socket" << std::endl;
}
}
newb_acceptor(const newb_acceptor& other) = delete;
......@@ -635,9 +638,16 @@ struct newb_acceptor : public newb_base, public caf::ref_counted {
if (accept_pol->manual_read) {
accept_pol->read_event(this);
} else {
using newb_ptr = first_argument_type<Fun>;
using newb_type = typename std::remove_pointer<newb_ptr>::type;
using message = typename newb_type::message_type;
network::native_socket sock;
policy::transport_ptr transport;
std::tie(sock, transport) = accept_pol->accept_event(this);
if (sock == network::invalid_native_socket) {
CAF_LOG_ERROR("failed to create socket for new endpoint");
return;
}
auto en = create_newb(sock, std::move(transport));
if (!en) {
io_error(network::operation::read, std::move(en.error()));
......@@ -645,8 +655,8 @@ struct newb_acceptor : public newb_base, public caf::ref_counted {
}
auto ptr = caf::actor_cast<caf::abstract_actor*>(*en);
CAF_ASSERT(ptr != nullptr);
auto& ref = dynamic_cast<newb<message_type>&>(*ptr);
accept_pol->init(ref);
auto& ref = dynamic_cast<newb<message>&>(*ptr);
accept_pol->init(this, ref);
}
}
......@@ -666,7 +676,7 @@ struct newb_acceptor : public newb_base, public caf::ref_counted {
return n;
}
policy::accept_ptr accept_pol;
policy::accept_ptr<message_type> accept_pol;
private:
Fun fun_;
......@@ -678,9 +688,9 @@ private:
template <class P, class F, class... Ts>
using acceptor_ptr = caf::intrusive_ptr<newb_acceptor<P, F, Ts...>>;
template <class Protocol, class Fun, class... Ts>
template <class Protocol, class Fun, class Message, class... Ts>
acceptor_ptr<Protocol, Fun, Ts...>
make_acceptor(actor_system& sys, Fun fun, policy::accept_ptr pol,
make_acceptor(actor_system& sys, Fun fun, policy::accept_ptr<Message> pol,
network::native_socket sockfd, Ts&&... xs) {
auto& dm = dynamic_cast<network::default_multiplexer&>(sys.middleman().backend());
auto res = make_counted<newb_acceptor<Protocol, Fun, Ts...>>(dm, sockfd,
......@@ -691,9 +701,9 @@ make_acceptor(actor_system& sys, Fun fun, policy::accept_ptr pol,
return res;
}
template <class Protocol, class F, class... Ts>
template <class Protocol, class F, class Message, class... Ts>
expected<caf::intrusive_ptr<newb_acceptor<Protocol, F, Ts...>>>
make_server(actor_system& sys, F fun, policy::accept_ptr pol,
make_server(actor_system& sys, F fun, policy::accept_ptr<Message> pol,
uint16_t port, const char* addr, bool reuse, Ts&&... xs) {
auto esock = pol->create_socket(port, addr, reuse);
if (!esock) {
......@@ -704,9 +714,9 @@ make_server(actor_system& sys, F fun, policy::accept_ptr pol,
std::forward<Ts>(xs)...);
}
template <class Protocol, class F>
template <class Protocol, class F, class Message>
expected<caf::intrusive_ptr<newb_acceptor<Protocol, F>>>
make_server(actor_system& sys, F fun, policy::accept_ptr pol,
make_server(actor_system& sys, F fun, policy::accept_ptr<Message> pol,
uint16_t port, const char* addr = nullptr, bool reuse = false) {
return make_server<Protocol, F>(sys, std::move(fun), std::move(pol), port,
addr, reuse);
......
......@@ -27,17 +27,23 @@ namespace io {
struct newb_base;
template<class Message>
struct newb;
} // namespace io
namespace policy {
template <class Message>
struct accept {
accept(bool manual_read = false)
: manual_read(manual_read) {
// nop
}
virtual ~accept();
virtual ~accept() {
// nop
}
virtual expected<io::network::native_socket>
create_socket(uint16_t port, const char* host, bool reuse = false) = 0;
......@@ -58,14 +64,15 @@ struct accept {
return none;
}
virtual void init(io::newb_base&) {
virtual void init(io::newb_base*, io::newb<Message>&) {
// nop
}
bool manual_read;
};
using accept_ptr = std::unique_ptr<accept>;
template <class Message>
using accept_ptr = std::unique_ptr<accept<Message>>;
} // namespace policy
} // namespace caf
......@@ -18,6 +18,7 @@
#pragma once
#include "caf/io/network/default_multiplexer.hpp"
#include "caf/io/network/native_socket.hpp"
#include "caf/policy/accept.hpp"
#include "caf/policy/transport.hpp"
......@@ -57,14 +58,28 @@ struct tcp_transport : public transport {
size_t written;
};
struct accept_tcp : public accept {
io::network::native_socket get_newb_socket(io::newb_base*);
template <class Message>
struct accept_tcp : public accept<Message> {
expected<io::network::native_socket>
create_socket(uint16_t port,const char* host,bool reuse = false) override;
create_socket(uint16_t port,const char* host,bool reuse = false) override {
return io::network::new_tcp_acceptor_impl(port, host, reuse);
}
std::pair<io::network::native_socket, transport_ptr>
accept_event(io::newb_base* parent) override;
void init(io::newb_base& n) override;
accept_event(io::newb_base* parent) override {
auto esock = io::network::accept_tcp_connection(get_newb_socket(parent));
if (!esock) {
return {io::network::invalid_native_socket, nullptr};
}
transport_ptr ptr{new tcp_transport};
return {*esock, std::move(ptr)};
}
void init(io::newb_base*, io::newb<Message>& spawned) override {
spawned.start();
}
};
template <class T>
......
......@@ -18,6 +18,7 @@
#pragma once
#include "caf/io/network/default_multiplexer.hpp"
#include "caf/io/network/ip_endpoint.hpp"
#include "caf/io/network/native_socket.hpp"
#include "caf/policy/accept.hpp"
......@@ -70,14 +71,34 @@ struct udp_transport : public transport {
io::network::ip_endpoint sender;
};
struct accept_udp : public accept {
template <class Message>
struct accept_udp : public accept<Message> {
expected<io::network::native_socket>
create_socket(uint16_t port, const char* host, bool reuse = false) override;
create_socket(uint16_t port, const char* host, bool reuse = false) override {
auto res = io::network::new_local_udp_endpoint_impl(port, host, reuse);
if (!res)
return std::move(res.error());
return (*res).first;
}
std::pair<io::network::native_socket, transport_ptr>
accept_event(io::newb_base*) override;
accept_event(io::newb_base*) override {
std::cout << "Accepting msg from UDP endpoint" << std::endl;
auto res = io::network::new_local_udp_endpoint_impl(0, nullptr, true);
if (!res) {
CAF_LOG_DEBUG("failed to create local endpoint" << CAF_ARG(res.error()));
return {io::network::invalid_native_socket, nullptr};
}
auto sock = std::move(res->first);
transport_ptr ptr{new udp_transport};
return {sock, std::move(ptr)};
}
void init(io::newb_base& n) override;
void init(io::newb_base* parent, io::newb<Message>& spawned) override {
spawned.trans->prepare_next_read(parent);
spawned.trans->read_some(parent, *spawned.proto.get());
spawned.start();
}
};
template <class T>
......
......@@ -954,6 +954,20 @@ new_local_udp_endpoint_impl(uint16_t port, const char* addr, bool reuse,
return std::make_pair(fd, proto);
}
expected<native_socket> accept_tcp_connection(native_socket sockfd) {
sockaddr_storage addr;
std::memset(&addr, 0, sizeof(addr));
socket_size_type addrlen = sizeof(addr);
auto result = ::accept(sockfd, reinterpret_cast<sockaddr*>(&addr),
&addrlen);
if (result == invalid_native_socket) {
auto err = last_socket_error_as_string();
CAF_LOG_ERROR("cannot accept tcp connection: " << CAF_ARG(err));
return sec::runtime_error;
}
return result;
}
} // namespace network
} // namespace io
} // namespace caf
......@@ -178,31 +178,8 @@ tcp_transport::connect(const std::string& host, uint16_t port,
return io::network::new_tcp_connection(host, port, preferred);
}
expected<io::network::native_socket>
accept_tcp::create_socket(uint16_t port, const char* host, bool reuse) {
return io::network::new_tcp_acceptor_impl(port, host, reuse);
}
std::pair<io::network::native_socket, transport_ptr>
accept_tcp::accept_event(io::newb_base* parent) {
using namespace io::network;
sockaddr_storage addr;
std::memset(&addr, 0, sizeof(addr));
socket_size_type addrlen = sizeof(addr);
auto result = ::accept(parent->fd(), reinterpret_cast<sockaddr*>(&addr),
&addrlen);
if (result == invalid_native_socket) {
auto err = last_socket_error();
if (!would_block_or_temporarily_unavailable(err)) {
return {invalid_native_socket, nullptr};
}
}
transport_ptr ptr{new tcp_transport};
return {result, std::move(ptr)};
}
void accept_tcp::init(io::newb_base& n) {
n.start();
io::network::native_socket get_newb_socket(io::newb_base* n) {
return n->fd();
}
} // namespace policy
......
......@@ -178,29 +178,5 @@ udp_transport::connect(const std::string& host, uint16_t port,
return res->first;
}
expected<io::network::native_socket>
accept_udp::create_socket(uint16_t port, const char* host, bool reuse) {
auto res = io::network::new_local_udp_endpoint_impl(port, host, reuse);
if (!res)
return std::move(res.error());
return (*res).first;
}
std::pair<io::network::native_socket, transport_ptr>
accept_udp::accept_event(io::newb_base*) {
auto res = io::network::new_local_udp_endpoint_impl(0, nullptr);
if (!res) {
CAF_LOG_DEBUG("failed to create local endpoint");
return {io::network::invalid_native_socket, nullptr};
}
auto sock = std::move(res->first);
transport_ptr ptr{new udp_transport};
return {sock, std::move(ptr)};
}
void accept_udp::init(io::newb_base& n) {
n.start();
}
} // namespace policy
} // namespace caf
......@@ -77,12 +77,6 @@ transport::connect(const std::string&, uint16_t,
return sec::bad_function_call;
}
// -- accept_policy ------------------------------------------------------------
accept::~accept() {
// nop
}
// -- protocol_policy_base -----------------------------------------------------
protocol_base::~protocol_base() {
......
......@@ -137,24 +137,6 @@ struct dummy_transport : public transport {
}
};
struct accept_policy_impl : public accept {
expected<native_socket> create_socket(uint16_t, const char*, bool) override {
return sec::bad_function_call;
}
std::pair<native_socket, transport_ptr>
accept_event(newb_base*) override {
auto esock = network::new_local_udp_endpoint_impl(0, nullptr);
CAF_REQUIRE(esock);
transport_ptr ptr{new dummy_transport};
return {esock->first, std::move(ptr)};
}
void init(newb_base& n) override {
n.handle_event(network::operation::read);
}
};
// -- config for controlled scheduling and multiplexing ------------------------
class config : public actor_system_config {
......
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