Commit ea6ab1f4 authored by Joseph Noir's avatar Joseph Noir

Let newbs keep a self-ref, add client factory fun

parent e0e0012e
...@@ -55,177 +55,61 @@ using io::network::last_socket_error_as_string; ...@@ -55,177 +55,61 @@ using io::network::last_socket_error_as_string;
namespace { namespace {
constexpr auto ipv4 = caf::io::network::protocol::ipv4;
//constexpr auto ipv6 = caf::io::network::protocol::ipv6;
auto addr_of(sockaddr_in& what) -> decltype(what.sin_addr)& {
return what.sin_addr;
}
auto family_of(sockaddr_in& what) -> decltype(what.sin_family)& {
return what.sin_family;
}
auto port_of(sockaddr_in& what) -> decltype(what.sin_port)& {
return what.sin_port;
}
auto addr_of(sockaddr_in6& what) -> decltype(what.sin6_addr)& {
return what.sin6_addr;
}
auto family_of(sockaddr_in6& what) -> decltype(what.sin6_family)& {
return what.sin6_family;
}
auto port_of(sockaddr_in6& what) -> decltype(what.sin6_port)& {
return what.sin6_port;
}
// -- atoms -------------------------------------------------------------------- // -- atoms --------------------------------------------------------------------
using ordering_atom = atom_constant<atom("ordering")>; using ordering_atom = atom_constant<atom("ordering")>;
using send_atom = atom_constant<atom("send")>; using send_atom = atom_constant<atom("send")>;
using quit_atom = atom_constant<atom("quit")>; using quit_atom = atom_constant<atom("quit")>;
using responder_atom = atom_constant<atom("responder")>;
// -- network code -------------------------------------------------------------
expected<void> set_inaddr_any(native_socket, sockaddr_in& sa) {
sa.sin_addr.s_addr = INADDR_ANY;
return unit;
}
expected<void> set_inaddr_any(native_socket fd, sockaddr_in6& sa) {
sa.sin6_addr = in6addr_any;
// also accept ipv4 requests on this socket
int off = 0;
CALL_CFUN(res, detail::cc_zero, "setsockopt",
setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY,
reinterpret_cast<io::network::setsockopt_ptr>(&off),
static_cast<io::network::socket_size_type>(sizeof(off))));
return unit;
}
template <int Family, int SockType = SOCK_STREAM>
expected<native_socket> new_ip_acceptor_impl(uint16_t port, const char* addr,
bool reuse_addr, bool any) {
static_assert(Family == AF_INET || Family == AF_INET6, "invalid family");
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr"));
CALL_CFUN(fd, detail::cc_valid_socket, "socket", socket(Family, SockType, 0));
// sguard closes the socket in case of exception
detail::socket_guard sguard{fd};
if (reuse_addr) {
int on = 1;
CALL_CFUN(tmp1, detail::cc_zero, "setsockopt",
setsockopt(fd, SOL_SOCKET, SO_REUSEADDR,
reinterpret_cast<io::network::setsockopt_ptr>(&on),
static_cast<io::network::socket_size_type>(sizeof(on))));
}
using sockaddr_type =
typename std::conditional<
Family == AF_INET,
sockaddr_in,
sockaddr_in6
>::type;
sockaddr_type sa;
memset(&sa, 0, sizeof(sockaddr_type));
family_of(sa) = Family;
if (any)
set_inaddr_any(fd, sa);
CALL_CFUN(tmp, detail::cc_one, "inet_pton",
inet_pton(Family, addr, &addr_of(sa)));
port_of(sa) = htons(port);
CALL_CFUN(res, detail::cc_zero, "bind",
bind(fd, reinterpret_cast<sockaddr*>(&sa),
static_cast<io::network::socket_size_type>(sizeof(sa))));
return sguard.release();
}
expected<native_socket> new_tcp_acceptor_impl(uint16_t port, const char* addr,
bool reuse_addr) {
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr"));
auto addrs = io::network::interfaces::server_address(port, addr);
auto addr_str = std::string{addr == nullptr ? "" : addr};
if (addrs.empty())
return make_error(sec::cannot_open_port, "No local interface available",
addr_str);
bool any = addr_str.empty() || addr_str == "::" || addr_str == "0.0.0.0";
auto fd = invalid_native_socket;
for (auto& elem : addrs) {
auto hostname = elem.first.c_str();
auto p = elem.second == ipv4
? new_ip_acceptor_impl<AF_INET>(port, hostname, reuse_addr, any)
: new_ip_acceptor_impl<AF_INET6>(port, hostname, reuse_addr, any);
if (!p) {
CAF_LOG_DEBUG(p.error());
continue;
}
fd = *p;
break;
}
if (fd == invalid_native_socket) {
CAF_LOG_WARNING("could not open tcp socket on:" << CAF_ARG(port)
<< CAF_ARG(addr_str));
return make_error(sec::cannot_open_port, "tcp socket creation failed",
port, addr_str);
}
detail::socket_guard sguard{fd};
CALL_CFUN(tmp2, detail::cc_zero, "listen", listen(fd, SOMAXCONN));
// ok, no errors so far
CAF_LOG_DEBUG(CAF_ARG(fd));
return sguard.release();
}
// -- tcp impls ---------------------------------------------------------------- // -- tcp impls ----------------------------------------------------------------
struct tcp_basp_header { struct basp_header {
uint32_t payload_len; uint32_t payload_len;
actor_id from; actor_id from;
actor_id to; actor_id to;
}; };
constexpr size_t tcp_basp_header_len = sizeof(uint32_t) + sizeof(actor_id) * 2; constexpr size_t basp_header_len = sizeof(uint32_t) + sizeof(actor_id) * 2;
template <class Inspector> template <class Inspector>
typename Inspector::result_type inspect(Inspector& fun, tcp_basp_header& hdr) { typename Inspector::result_type inspect(Inspector& fun, basp_header& hdr) {
return fun(meta::type_name("tcp_basp_header"), return fun(meta::type_name("tcp_basp_header"),
hdr.payload_len, hdr.from, hdr.to); hdr.payload_len, hdr.from, hdr.to);
} }
struct new_tcp_basp_message { struct new_basp_message {
tcp_basp_header header; basp_header header;
char* payload; char* payload;
size_t payload_len; size_t payload_len;
}; };
template <class Inspector> template <class Inspector>
typename Inspector::result_type inspect(Inspector& fun, typename Inspector::result_type inspect(Inspector& fun,
new_tcp_basp_message& msg) { new_basp_message& msg) {
return fun(meta::type_name("new_tcp_basp_message"), msg.header, return fun(meta::type_name("tcp_new_basp_message"), msg.header,
msg.payload_len); msg.payload_len);
} }
struct tcp_basp { struct basp {
static constexpr size_t header_size = sizeof(tcp_basp_header); static constexpr size_t header_size = basp_header_len;
using message_type = new_tcp_basp_message; using message_type = new_basp_message;
using result_type = optional<message_type>; using result_type = optional<message_type>;
io::network::newb<message_type>* parent; io::network::newb<message_type>* parent;
message_type msg; message_type msg;
bool expecting_header = true; bool expecting_header = true;
tcp_basp(io::network::newb<message_type>* parent) : parent(parent) { basp(io::network::newb<message_type>* parent) : parent(parent) {
// nop // TODO: this is dangerous ...
// Maybe we need an init function that is called with `start()`?
parent->configure_read(io::receive_policy::exactly(basp_header_len));
} }
error read_header(char* bytes, size_t count) { error read_header(char* bytes, size_t count) {
if (count < tcp_basp_header_len) { if (count < basp_header_len)
CAF_LOG_DEBUG("buffer contains " << count << " bytes of expected "
<< tcp_basp_header_len);
return sec::unexpected_message; return sec::unexpected_message;
}
binary_deserializer bd{&parent->backend(), bytes, count}; binary_deserializer bd{&parent->backend(), bytes, count};
bd(msg.header); bd(msg.header);
CAF_LOG_DEBUG("read header " << CAF_ARG(msg.header));
size_t size = static_cast<size_t>(msg.header.payload_len); size_t size = static_cast<size_t>(msg.header.payload_len);
parent->configure_read(io::receive_policy::exactly(size)); parent->configure_read(io::receive_policy::exactly(size));
expecting_header = false; expecting_header = false;
...@@ -242,7 +126,7 @@ struct tcp_basp { ...@@ -242,7 +126,7 @@ struct tcp_basp {
msg.payload_len = msg.header.payload_len; msg.payload_len = msg.header.payload_len;
parent->handle(msg); parent->handle(msg);
expecting_header = true; expecting_header = true;
parent->configure_read(io::receive_policy::exactly(tcp_basp_header_len)); parent->configure_read(io::receive_policy::exactly(basp_header_len));
return none; return none;
} }
...@@ -273,8 +157,8 @@ struct tcp_basp { ...@@ -273,8 +157,8 @@ struct tcp_basp {
} }
}; };
struct tcp_transport_policy : public io::network::transport_policy { struct tcp_transport : public io::network::transport_policy {
tcp_transport_policy() tcp_transport()
: read_threshold{0}, : read_threshold{0},
collected{0}, collected{0},
maximum{0}, maximum{0},
...@@ -285,6 +169,7 @@ struct tcp_transport_policy : public io::network::transport_policy { ...@@ -285,6 +169,7 @@ struct tcp_transport_policy : public io::network::transport_policy {
error read_some(io::network::event_handler* parent) override { error read_some(io::network::event_handler* parent) override {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
std::cerr << "read some called" << std::endl;
size_t len = receive_buffer.size() - collected; size_t len = receive_buffer.size() - collected;
receive_buffer.resize(len); receive_buffer.resize(len);
void* buf = receive_buffer.data() + collected; void* buf = receive_buffer.data() + collected;
...@@ -292,6 +177,7 @@ struct tcp_transport_policy : public io::network::transport_policy { ...@@ -292,6 +177,7 @@ struct tcp_transport_policy : public io::network::transport_policy {
reinterpret_cast<io::network::socket_recv_ptr>(buf), reinterpret_cast<io::network::socket_recv_ptr>(buf),
len, io::network::no_sigpipe_io_flag); len, io::network::no_sigpipe_io_flag);
if (io::network::is_error(sres, true) || sres == 0) { if (io::network::is_error(sres, true) || sres == 0) {
std::cerr << "read some error" << std::endl;
// recv returns 0 when the peer has performed an orderly shutdown // recv returns 0 when the peer has performed an orderly shutdown
return sec::runtime_error; return sec::runtime_error;
} }
...@@ -380,6 +266,15 @@ struct tcp_transport_policy : public io::network::transport_policy { ...@@ -380,6 +266,15 @@ struct tcp_transport_policy : public io::network::transport_policy {
} }
} }
expected<native_socket>
connect(const std::string& host, uint16_t port,
optional<io::network::protocol::network> preferred = none) override {
auto res = io::network::new_tcp_connection(host, port, preferred);
if (!res)
std::cerr << "failed to create new TCP connection" << std::endl;
return res;
}
// State for reading. // State for reading.
size_t read_threshold; size_t read_threshold;
size_t collected; size_t collected;
...@@ -392,11 +287,11 @@ struct tcp_transport_policy : public io::network::transport_policy { ...@@ -392,11 +287,11 @@ struct tcp_transport_policy : public io::network::transport_policy {
}; };
template <class T> template <class T>
struct tcp_protocol_policy struct tcp_protocol
: public io::network::protocol_policy<typename T::message_type> { : public io::network::protocol_policy<typename T::message_type> {
T impl; T impl;
tcp_protocol_policy(io::network::newb<typename T::message_type>* parent) tcp_protocol(io::network::newb<typename T::message_type>* parent)
: impl(parent) { : impl(parent) {
// nop // nop
} }
...@@ -420,14 +315,21 @@ struct tcp_protocol_policy ...@@ -420,14 +315,21 @@ struct tcp_protocol_policy
} }
}; };
struct tcp_basp_newb : io::network::newb<new_tcp_basp_message> { struct basp_newb : public io::network::newb<new_basp_message> {
tcp_basp_newb(caf::actor_config& cfg, default_multiplexer& dm, basp_newb(caf::actor_config& cfg, default_multiplexer& dm,
native_socket sockfd) native_socket sockfd)
: newb<new_tcp_basp_message>(cfg, dm, sockfd) { : newb<new_basp_message>(cfg, dm, sockfd) {
// nop // nop
CAF_LOG_TRACE("");
std::cerr << "constructing newb" << std::endl;
}
~basp_newb() {
std::cerr << "terminating newb" << std::endl;
CAF_LOG_TRACE("");
} }
void handle(new_tcp_basp_message& msg) override { void handle(new_basp_message& msg) override {
CAF_PUSH_AID_FROM_PTR(this); CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
std::string res; std::string res;
...@@ -447,7 +349,7 @@ struct tcp_basp_newb : io::network::newb<new_tcp_basp_message> { ...@@ -447,7 +349,7 @@ struct tcp_basp_newb : io::network::newb<new_tcp_basp_message> {
[=](send_atom, actor_id sender, actor_id receiver, std::string payload) { [=](send_atom, actor_id sender, actor_id receiver, std::string payload) {
auto hw = caf::make_callback([&](io::network::byte_buffer& buf) -> error { auto hw = caf::make_callback([&](io::network::byte_buffer& buf) -> error {
binary_serializer bs(&backend(), buf); binary_serializer bs(&backend(), buf);
bs(tcp_basp_header{0, sender, receiver}); bs(basp_header{0, sender, receiver});
return none; return none;
}); });
auto whdl = wr_buf(&hw); auto whdl = wr_buf(&hw);
...@@ -456,7 +358,13 @@ struct tcp_basp_newb : io::network::newb<new_tcp_basp_message> { ...@@ -456,7 +358,13 @@ struct tcp_basp_newb : io::network::newb<new_tcp_basp_message> {
binary_serializer bs(&backend(), *whdl.buf); binary_serializer bs(&backend(), *whdl.buf);
bs(payload); bs(payload);
}, },
[=](responder_atom, actor r) {
aout(this) << "got responder assigned" << std::endl;
responder = r;
send(r, this);
},
[=](quit_atom) { [=](quit_atom) {
aout(this) << "got quit message" << std::endl;
// Remove from multiplexer loop. // Remove from multiplexer loop.
stop(); stop();
// Quit actor. // Quit actor.
...@@ -469,10 +377,14 @@ struct tcp_basp_newb : io::network::newb<new_tcp_basp_message> { ...@@ -469,10 +377,14 @@ struct tcp_basp_newb : io::network::newb<new_tcp_basp_message> {
}; };
struct tcp_accept_policy struct accept_tcp : public io::network::accept_policy<new_basp_message> {
: public io::network::accept_policy<new_tcp_basp_message> { expected<native_socket> create_socket(uint16_t port, const char* host,
virtual std::pair<native_socket, io::network::transport_policy_ptr> bool reuse = false) override {
accept(io::network::event_handler* parent) { return io::network::new_tcp_acceptor_impl(port, host, reuse);
}
std::pair<native_socket, io::network::transport_policy_ptr>
accept(io::network::event_handler* parent) override {
using namespace io::network; using namespace io::network;
sockaddr_storage addr; sockaddr_storage addr;
std::memset(&addr, 0, sizeof(addr)); std::memset(&addr, 0, sizeof(addr));
...@@ -485,45 +397,48 @@ struct tcp_accept_policy ...@@ -485,45 +397,48 @@ struct tcp_accept_policy
return {invalid_native_socket, nullptr}; return {invalid_native_socket, nullptr};
} }
} }
transport_policy_ptr ptr{new tcp_transport_policy}; std::cerr << "accepted connection" << std::endl;
transport_policy_ptr ptr{new tcp_transport};
return {result, std::move(ptr)}; return {result, std::move(ptr)};
} }
virtual void init(io::network::newb<new_tcp_basp_message>& n) { void init(io::network::newb<new_basp_message>& n) override {
n.start(); n.start();
} }
}; };
template <class ProtocolPolicy> template <class ProtocolPolicy>
struct tcp_basp_acceptor struct tcp_acceptor
: public io::network::newb_acceptor<typename ProtocolPolicy::message_type> { : public io::network::newb_acceptor<typename ProtocolPolicy::message_type> {
using super = io::network::newb_acceptor<typename ProtocolPolicy::message_type>; using super = io::network::newb_acceptor<typename ProtocolPolicy::message_type>;
tcp_acceptor(default_multiplexer& dm, native_socket sockfd)
static expected<native_socket> create_socket(uint16_t port, const char* host,
bool reuse = false) {
return new_tcp_acceptor_impl(port, host, reuse);
}
tcp_basp_acceptor(default_multiplexer& dm, native_socket sockfd)
: super(dm, sockfd) { : super(dm, sockfd) {
CAF_LOG_TRACE("");
std::cerr << "constructing newb acceptor" << std::endl;
// nop // nop
} }
~tcp_acceptor() {
CAF_LOG_TRACE("");
std::cerr << "terminating newb acceptor" << std::endl;
}
expected<actor> create_newb(native_socket sockfd, expected<actor> create_newb(native_socket sockfd,
io::network::transport_policy_ptr pol) override { io::network::transport_policy_ptr pol) override {
CAF_LOG_DEBUG("creating new basp tcp newb"); CAF_LOG_TRACE(CAF_ARG(sockfd));
auto n = io::network::make_newb<tcp_basp_newb>(this->backend().system(), std::cerr << "acceptor creating new newb" << std::endl;
sockfd); auto n = io::network::make_newb<basp_newb>(this->backend().system(), sockfd);
auto ptr = caf::actor_cast<caf::abstract_actor*>(n); auto ptr = caf::actor_cast<caf::abstract_actor*>(n);
if (ptr == nullptr) if (ptr == nullptr)
return sec::runtime_error; return sec::runtime_error;
auto& ref = dynamic_cast<tcp_basp_newb&>(*ptr); auto& ref = dynamic_cast<basp_newb&>(*ptr);
// TODO: Transport has to be assigned before protocol ... which sucks.
// (basp protocol calls configure read which accesses the transport.)
ref.transport = std::move(pol); ref.transport = std::move(pol);
ref.protocol.reset(new ProtocolPolicy(&ref)); ref.protocol.reset(new ProtocolPolicy(&ref));
ref.responder = responder; ref.responder = responder;
// This should happen somewhere else? // TODO: Just a workaround.
ref.configure_read(io::receive_policy::exactly(tcp_basp_header_len));
anon_send(responder, n); anon_send(responder, n);
return n; return n;
} }
...@@ -532,126 +447,61 @@ struct tcp_basp_acceptor ...@@ -532,126 +447,61 @@ struct tcp_basp_acceptor
}; };
struct tcp_test_broker_state { struct tcp_test_broker_state {
tcp_basp_header hdr; basp_header hdr;
bool expecting_header = true; bool expecting_header = true;
}; };
void caf_main(actor_system& sys, const actor_system_config&) { void caf_main(actor_system& sys, const actor_system_config&) {
using tcp_protocol_policy_t = tcp_protocol_policy<tcp_basp>; using acceptor_t = tcp_acceptor<tcp_protocol<basp>>;
using tcp_accept_policy_t = tcp_accept_policy; using io::network::make_server_newb;
using tcp_newb_acceptor_t = tcp_basp_acceptor<tcp_protocol_policy_t>; using io::network::make_client_newb;
using tcp_transport_policy_t = tcp_transport_policy;
const char* host = "localhost"; const char* host = "localhost";
const uint16_t port = 12345; const uint16_t port = 12345;
scoped_actor self{sys};
scoped_actor main_actor{sys}; auto running = [=](event_based_actor* self, std::string name, actor , actor b) -> behavior {
actor newb_actor;
auto testing = [&](io::stateful_broker<tcp_test_broker_state>* self,
io::connection_handle hdl, actor) -> behavior {
CAF_ASSERT(hdl != io::invalid_connection_handle);
self->configure_read(hdl, io::receive_policy::exactly(tcp_basp_header_len));
self->state.expecting_header = true;
return { return {
[=](send_atom, std::string str) { [=](std::string str) {
CAF_LOG_DEBUG("sending '" << str << "'"); aout(self) << "[" << name << "] received '" << str << "'" << std::endl;
io::network::byte_buffer buf;
binary_serializer bs(self->system(), buf);
tcp_basp_header hdr{0, 1, 2};
bs(hdr);
auto header_len = buf.size();
CAF_ASSERT(header_len == tcp_basp_header_len);
bs(str);
hdr.payload_len = static_cast<uint32_t>(buf.size() - header_len);
stream_serializer<charbuf> out{self->system(), buf.data(),
sizeof(hdr.payload_len)};
out(hdr.payload_len);
CAF_LOG_DEBUG("header len: " << header_len
<< ", packet_len: " << buf.size()
<< ", header: " << to_string(hdr));
self->write(hdl, buf.size(), buf.data());
self->flush(hdl);
}, },
[=](quit_atom) { [=](send_atom, std::string str) {
CAF_LOG_DEBUG("test broker shutting down"); aout(self) << "[" << name << "] sending '" << str << "'" << std::endl;
self->quit(); self->send(b, send_atom::value, self->id(), actor_id{}, str);
}, },
[=](io::new_data_msg& msg) {
auto& s = self->state;
size_t next_len = tcp_basp_header_len;
binary_deserializer bd(self->system(), msg.buf);
if (s.expecting_header) {
bd(s.hdr);
next_len = s.hdr.payload_len;
s.expecting_header = false;
} else {
std::string str;
bd(str);
CAF_LOG_DEBUG("received '" << str << "'");
std::reverse(std::begin(str), std::end(str));
io::network::byte_buffer buf;
binary_serializer bs(self->system(), buf);
tcp_basp_header hdr{0, 1, 2};
bs(hdr);
auto header_len = buf.size();
CAF_ASSERT(header_len == tcp_basp_header_len);
bs(str);
hdr.payload_len = static_cast<uint32_t>(buf.size() - header_len);
stream_serializer<charbuf> out{self->system(), buf.data(),
sizeof(hdr.payload_len)};
out(hdr.payload_len);
CAF_LOG_DEBUG("header len: " << header_len
<< ", packet_len: " << buf.size()
<< ", header: " << to_string(hdr));
self->write(hdl, buf.size(), buf.data());
self->flush(hdl);
// self->send(m, quit_atom::value);
}
self->configure_read(msg.handle, io::receive_policy::exactly(next_len));
}
}; };
}; };
auto helper_actor = sys.spawn([&](event_based_actor* self, actor m) -> behavior { auto init = [=](event_based_actor* self, std::string name, actor m) -> behavior {
self->set_default_handler(skip);
return { return {
[=](const std::string& str) { [=](actor b) {
CAF_LOG_DEBUG("received '" << str << "'"); aout(self) << "[" << name << "] got broker, let's do this" << std::endl;
self->send(m, quit_atom::value); self->become(running(self, name, m, b));
}, self->set_default_handler(print_and_drop);
[=](actor a) {
CAF_LOG_DEBUG("got new newb handle");
self->send(m, a);
},
[=](quit_atom) {
CAF_LOG_DEBUG("helper shutting down");
self->quit();
} }
}; };
}, main_actor); };
CAF_LOG_DEBUG("creating new acceptor");
auto newb_acceptor_ptr auto server_helper = sys.spawn(init, "s", self);
= io::network::make_newb_acceptor<tcp_newb_acceptor_t, tcp_accept_policy_t>(sys, port); auto client_helper = sys.spawn(init, "c", self);
dynamic_cast<tcp_newb_acceptor_t*>(newb_acceptor_ptr.get())->responder
= helper_actor; aout(self) << "creating new server" << std::endl;
CAF_LOG_DEBUG("connecting from 'old-style' broker"); auto server_ptr = make_server_newb<acceptor_t, accept_tcp>(sys, port, nullptr, true);
auto exp = sys.middleman().spawn_client(testing, host, port, main_actor); server_ptr->responder = server_helper;
CAF_ASSERT(exp);
auto test_broker = std::move(*exp); aout(self) << "creating new client" << std::endl;
main_actor->receive( auto client = make_client_newb<basp_newb, tcp_transport, tcp_protocol<basp>>(sys, host, port);
[&](actor a) { self->send(client, responder_atom::value, client_helper);
newb_actor = a;
} self->send(client_helper, send_atom::value, "hallo");
); self->send(server_helper, send_atom::value, "hallo");
CAF_LOG_DEBUG("sending message to newb");
main_actor->send(test_broker, send_atom::value, "hello world"); self->receive(
std::this_thread::sleep_for(std::chrono::seconds(1)); [&](quit_atom) {
main_actor->receive( aout(self) << "check" << std::endl;
[](quit_atom) {
CAF_LOG_DEBUG("check");
} }
); );
CAF_LOG_DEBUG("sending message from newb");
main_actor->send(newb_actor, send_atom::value, actor_id{3}, actor_id{4}, /*
"dlrow olleh");
main_actor->receive( main_actor->receive(
[](quit_atom) { [](quit_atom) {
CAF_LOG_DEBUG("check"); CAF_LOG_DEBUG("check");
...@@ -664,6 +514,7 @@ void caf_main(actor_system& sys, const actor_system_config&) { ...@@ -664,6 +514,7 @@ void caf_main(actor_system& sys, const actor_system_config&) {
anon_send(test_broker, quit_atom::value); anon_send(test_broker, quit_atom::value);
sys.await_all_actors_done(); sys.await_all_actors_done();
CAF_LOG_DEBUG("done"); CAF_LOG_DEBUG("done");
*/
} }
} // namespace anonymous } // namespace anonymous
......
...@@ -55,124 +55,12 @@ using io::network::last_socket_error_as_string; ...@@ -55,124 +55,12 @@ using io::network::last_socket_error_as_string;
namespace { namespace {
constexpr auto ipv4 = caf::io::network::protocol::ipv4;
//constexpr auto ipv6 = caf::io::network::protocol::ipv6;
auto addr_of(sockaddr_in& what) -> decltype(what.sin_addr)& {
return what.sin_addr;
}
auto family_of(sockaddr_in& what) -> decltype(what.sin_family)& {
return what.sin_family;
}
auto port_of(sockaddr_in& what) -> decltype(what.sin_port)& {
return what.sin_port;
}
auto addr_of(sockaddr_in6& what) -> decltype(what.sin6_addr)& {
return what.sin6_addr;
}
auto family_of(sockaddr_in6& what) -> decltype(what.sin6_family)& {
return what.sin6_family;
}
auto port_of(sockaddr_in6& what) -> decltype(what.sin6_port)& {
return what.sin6_port;
}
// -- atoms -------------------------------------------------------------------- // -- atoms --------------------------------------------------------------------
using ordering_atom = atom_constant<atom("ordering")>; using ordering_atom = atom_constant<atom("ordering")>;
using send_atom = atom_constant<atom("send")>; using send_atom = atom_constant<atom("send")>;
using quit_atom = atom_constant<atom("quit")>; using quit_atom = atom_constant<atom("quit")>;
// -- network code -------------------------------------------------------------
expected<void> set_inaddr_any(native_socket, sockaddr_in& sa) {
sa.sin_addr.s_addr = INADDR_ANY;
return unit;
}
expected<void> set_inaddr_any(native_socket fd, sockaddr_in6& sa) {
sa.sin6_addr = in6addr_any;
// also accept ipv4 requests on this socket
int off = 0;
CALL_CFUN(res, detail::cc_zero, "setsockopt",
setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY,
reinterpret_cast<io::network::setsockopt_ptr>(&off),
static_cast<io::network::socket_size_type>(sizeof(off))));
return unit;
}
template <int Family, int SockType = SOCK_STREAM>
expected<native_socket> new_ip_acceptor_impl(uint16_t port, const char* addr,
bool reuse_addr, bool any) {
static_assert(Family == AF_INET || Family == AF_INET6, "invalid family");
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr"));
CALL_CFUN(fd, detail::cc_valid_socket, "socket", socket(Family, SockType, 0));
// sguard closes the socket in case of exception
detail::socket_guard sguard{fd};
if (reuse_addr) {
int on = 1;
CALL_CFUN(tmp1, detail::cc_zero, "setsockopt",
setsockopt(fd, SOL_SOCKET, SO_REUSEADDR,
reinterpret_cast<io::network::setsockopt_ptr>(&on),
static_cast<io::network::socket_size_type>(sizeof(on))));
}
using sockaddr_type =
typename std::conditional<
Family == AF_INET,
sockaddr_in,
sockaddr_in6
>::type;
sockaddr_type sa;
memset(&sa, 0, sizeof(sockaddr_type));
family_of(sa) = Family;
if (any)
set_inaddr_any(fd, sa);
CALL_CFUN(tmp, detail::cc_one, "inet_pton",
inet_pton(Family, addr, &addr_of(sa)));
port_of(sa) = htons(port);
CALL_CFUN(res, detail::cc_zero, "bind",
bind(fd, reinterpret_cast<sockaddr*>(&sa),
static_cast<io::network::socket_size_type>(sizeof(sa))));
return sguard.release();
}
expected<native_socket> new_udp_endpoint_impl(uint16_t port, const char* addr,
bool reuse) {
CAF_LOG_TRACE(CAF_ARG(port) << ", addr = " << (addr ? addr : "nullptr"));
auto addrs = io::network::interfaces::server_address(port, addr);
auto addr_str = std::string{addr == nullptr ? "" : addr};
if (addrs.empty())
return make_error(sec::cannot_open_port, "No local interface available",
addr_str);
bool any = addr_str.empty() || addr_str == "::" || addr_str == "0.0.0.0";
auto fd = invalid_native_socket;
for (auto& elem : addrs) {
auto host = elem.first.c_str();
auto p = elem.second == ipv4
? new_ip_acceptor_impl<AF_INET, SOCK_DGRAM>(port, host, reuse, any)
: new_ip_acceptor_impl<AF_INET6, SOCK_DGRAM>(port, host, reuse, any);
if (!p) {
CAF_LOG_DEBUG(p.error());
continue;
}
fd = *p;
break;
}
if (fd == invalid_native_socket) {
CAF_LOG_WARNING("could not open udp socket on:" << CAF_ARG(port)
<< CAF_ARG(addr_str));
return make_error(sec::cannot_open_port, "udp socket creation failed",
port, addr_str);
}
CAF_LOG_DEBUG(CAF_ARG(fd));
return fd;
}
// -- udp impls ---------------------------------------------------------------- // -- udp impls ----------------------------------------------------------------
struct udp_basp_header { struct udp_basp_header {
...@@ -227,24 +115,24 @@ struct udp_basp { ...@@ -227,24 +115,24 @@ struct udp_basp {
} }
error read(char* bytes, size_t count) { error read(char* bytes, size_t count) {
CAF_LOG_DEBUG("reading basp udp header"); std::cerr << "reading basp udp header" << std::endl;
// Read header. // Read header.
if (count < udp_basp_header_len) { if (count < udp_basp_header_len) {
CAF_LOG_DEBUG("not enought bytes for basp header"); std::cerr << "not enought bytes for basp header" << std::endl;
CAF_LOG_DEBUG("buffer contains " << count << " bytes of expected " std::cerr << "buffer contains " << count << " bytes of expected "
<< udp_basp_header_len); << udp_basp_header_len << std::endl;
return sec::unexpected_message; return sec::unexpected_message;
} }
binary_deserializer bd{&parent->backend(), bytes, count}; binary_deserializer bd{&parent->backend(), bytes, count};
bd(msg.header); bd(msg.header);
CAF_LOG_DEBUG("read basp header " << CAF_ARG(msg.header)); std::cerr << "read basp header " << to_string(msg.header) << std::endl;
size_t payload_len = static_cast<size_t>(msg.header.payload_len); size_t payload_len = static_cast<size_t>(msg.header.payload_len);
// Read payload. // Read payload.
auto remaining = count - udp_basp_header_len; auto remaining = count - udp_basp_header_len;
// TODO: Could be `!=` ? // TODO: Could be `!=` ?
if (remaining < payload_len) { if (remaining < payload_len) {
CAF_LOG_DEBUG("only " << remaining << " bytes remaining of expected " std::cerr << "only " << remaining << " bytes remaining of expected "
<< msg.header.payload_len); << msg.header.payload_len << std::endl;
return sec::unexpected_message; return sec::unexpected_message;
} }
msg.payload = bytes + udp_basp_header_len; msg.payload = bytes + udp_basp_header_len;
...@@ -322,7 +210,7 @@ struct udp_ordering { ...@@ -322,7 +210,7 @@ struct udp_ordering {
udp_ordering_header hdr; udp_ordering_header hdr;
binary_deserializer bd(&parent->backend(), bytes, count); binary_deserializer bd(&parent->backend(), bytes, count);
bd(hdr); bd(hdr);
CAF_LOG_DEBUG("read udp ordering header: " << to_string(hdr)); std::cerr << "read udp ordering header: " << to_string(hdr) << std::endl;
// TODO: Use the comparison function from BASP instance. // TODO: Use the comparison function from BASP instance.
if (hdr.seq == seq_read) { if (hdr.seq == seq_read) {
seq_read += 1; seq_read += 1;
...@@ -388,7 +276,7 @@ struct udp_transport_policy : public io::network::transport_policy { ...@@ -388,7 +276,7 @@ struct udp_transport_policy : public io::network::transport_policy {
return sec::runtime_error; return sec::runtime_error;
} else if (io::network::would_block_or_temporarily_unavailable( } else if (io::network::would_block_or_temporarily_unavailable(
io::network::last_socket_error())) { io::network::last_socket_error())) {
CAF_LOG_DEBUG("try later"); std::cerr << "try later" << std::endl;
return sec::end_of_stream; return sec::end_of_stream;
} }
if (sres == 0) if (sres == 0)
...@@ -434,6 +322,7 @@ struct udp_transport_policy : public io::network::transport_policy { ...@@ -434,6 +322,7 @@ struct udp_transport_policy : public io::network::transport_policy {
send_sizes.pop_front(); send_sizes.pop_front();
written += (sres > 0) ? static_cast<size_t>(sres) : 0; written += (sres > 0) ? static_cast<size_t>(sres) : 0;
auto remaining = send_buffer.size() - written; auto remaining = send_buffer.size() - written;
std::cerr << "wrote '" << written << "' got '" << remaining << "' bytes left to write" << std::endl;
if (remaining == 0) if (remaining == 0)
prepare_next_write(parent); prepare_next_write(parent);
return none; return none;
...@@ -463,6 +352,7 @@ struct udp_transport_policy : public io::network::transport_policy { ...@@ -463,6 +352,7 @@ struct udp_transport_policy : public io::network::transport_policy {
auto chunk_size = offline_buffer.size() - offline_sum; auto chunk_size = offline_buffer.size() - offline_sum;
offline_sizes.push_back(chunk_size); offline_sizes.push_back(chunk_size);
offline_sum += chunk_size; offline_sum += chunk_size;
std::cerr << "adding chunk '" << chunk_size << "' up to total of '" << offline_sum << std::endl;
} }
return offline_buffer; return offline_buffer;
} }
...@@ -478,6 +368,16 @@ struct udp_transport_policy : public io::network::transport_policy { ...@@ -478,6 +368,16 @@ struct udp_transport_policy : public io::network::transport_policy {
} }
} }
expected<native_socket>
connect(const std::string& host, uint16_t port,
optional<io::network::protocol::network> preferred = none) override {
auto res = io::network::new_remote_udp_endpoint_impl(host, port, preferred);
if (!res)
return std::move(res.error());
endpoint = res->second;
return res->first;
}
// State for reading. // State for reading.
size_t maximum; size_t maximum;
bool first_message; bool first_message;
...@@ -527,9 +427,14 @@ struct udp_basp_newb : public io::network::newb<new_udp_basp_message> { ...@@ -527,9 +427,14 @@ struct udp_basp_newb : public io::network::newb<new_udp_basp_message> {
udp_basp_newb(caf::actor_config& cfg, default_multiplexer& dm, udp_basp_newb(caf::actor_config& cfg, default_multiplexer& dm,
native_socket sockfd) native_socket sockfd)
: newb<new_udp_basp_message>(cfg, dm, sockfd) { : newb<new_udp_basp_message>(cfg, dm, sockfd) {
std::cerr << "constructing udp newb" << std::endl;
// nop // nop
} }
~udp_basp_newb() {
std::cerr << "terminating udp newb" << std::endl;
}
void handle(new_udp_basp_message& msg) override { void handle(new_udp_basp_message& msg) override {
CAF_PUSH_AID_FROM_PTR(this); CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
...@@ -574,14 +479,22 @@ struct udp_basp_newb : public io::network::newb<new_udp_basp_message> { ...@@ -574,14 +479,22 @@ struct udp_basp_newb : public io::network::newb<new_udp_basp_message> {
struct udp_accept_policy struct udp_accept_policy
: public io::network::accept_policy<new_udp_basp_message> { : public io::network::accept_policy<new_udp_basp_message> {
expected<native_socket> 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<native_socket, io::network::transport_policy_ptr> std::pair<native_socket, io::network::transport_policy_ptr>
accept(io::network::event_handler*) override { accept(io::network::event_handler*) override {
auto esock = new_udp_endpoint_impl(0, nullptr, false); auto res = io::network::new_local_udp_endpoint_impl(0, nullptr);
if (!esock) if (!res)
return {invalid_native_socket, nullptr}; return {invalid_native_socket, nullptr};
auto result = std::move(*esock); auto sock = std::move(res->second);
io::network::transport_policy_ptr ptr{new udp_transport_policy}; io::network::transport_policy_ptr ptr{new udp_transport_policy};
return {result, std::move(ptr)}; return {sock, std::move(ptr)};
} }
void init(io::network::newb<new_udp_basp_message>& n) override { void init(io::network::newb<new_udp_basp_message>& n) override {
...@@ -599,14 +512,9 @@ struct udp_basp_acceptor ...@@ -599,14 +512,9 @@ struct udp_basp_acceptor
// nop // nop
} }
static expected<native_socket> create_socket(uint16_t port, const char* host,
bool reuse = false) {
return new_udp_endpoint_impl(port, host, reuse);
}
expected<actor> create_newb(native_socket sockfd, expected<actor> create_newb(native_socket sockfd,
io::network::transport_policy_ptr pol) override { io::network::transport_policy_ptr pol) override {
CAF_LOG_DEBUG("creating new basp udp newb"); std::cerr << "creating new basp udp newb" << std::endl;
auto n = io::network::make_newb<udp_basp_newb>(this->backend().system(), auto n = io::network::make_newb<udp_basp_newb>(this->backend().system(),
sockfd); sockfd);
auto ptr = caf::actor_cast<caf::abstract_actor*>(n); auto ptr = caf::actor_cast<caf::abstract_actor*>(n);
...@@ -651,7 +559,7 @@ void caf_main(actor_system& sys, const actor_system_config&) { ...@@ -651,7 +559,7 @@ void caf_main(actor_system& sys, const actor_system_config&) {
self->state.hdl = std::move(*ehdl); self->state.hdl = std::move(*ehdl);
return { return {
[=](send_atom, std::string str) { [=](send_atom, std::string str) {
CAF_LOG_DEBUG("sending '" << str << "'"); std::cerr << "sending '" << str << "'" << std::endl;
io::network::byte_buffer buf; io::network::byte_buffer buf;
udp_ordering_header ohdr{0}; udp_ordering_header ohdr{0};
udp_basp_header bhdr{0, 1, 2}; udp_basp_header bhdr{0, 1, 2};
...@@ -668,15 +576,15 @@ void caf_main(actor_system& sys, const actor_system_config&) { ...@@ -668,15 +576,15 @@ void caf_main(actor_system& sys, const actor_system_config&) {
buf.data() + ordering_header_len, buf.data() + ordering_header_len,
sizeof(bhdr.payload_len)}; sizeof(bhdr.payload_len)};
out(bhdr.payload_len); out(bhdr.payload_len);
CAF_LOG_DEBUG("header len: " << header_len std::cerr << "header len: " << header_len
<< ", packet_len: " << buf.size() << ", packet_len: " << buf.size()
<< ", ordering header: " << to_string(ohdr) << ", ordering header: " << to_string(ohdr)
<< ", basp header: " << to_string(bhdr)); << ", basp header: " << to_string(bhdr) << std::endl;
self->enqueue_datagram(self->state.hdl, std::move(buf)); self->enqueue_datagram(self->state.hdl, std::move(buf));
self->flush(self->state.hdl); self->flush(self->state.hdl);
}, },
[=](quit_atom) { [=](quit_atom) {
CAF_LOG_DEBUG("test broker shutting down"); std::cerr << "test broker shutting down" << std::endl;
self->quit(); self->quit();
}, },
[=](io::new_datagram_msg& msg) { [=](io::new_datagram_msg& msg) {
...@@ -687,7 +595,7 @@ void caf_main(actor_system& sys, const actor_system_config&) { ...@@ -687,7 +595,7 @@ void caf_main(actor_system& sys, const actor_system_config&) {
bd(ohdr); bd(ohdr);
bd(bhdr); bd(bhdr);
bd(str); bd(str);
CAF_LOG_DEBUG("received '" << str << "'"); std::cerr << "received '" << str << "'" << std::endl;
self->send(m, quit_atom::value); self->send(m, quit_atom::value);
} }
}; };
...@@ -695,26 +603,26 @@ void caf_main(actor_system& sys, const actor_system_config&) { ...@@ -695,26 +603,26 @@ void caf_main(actor_system& sys, const actor_system_config&) {
auto helper_actor = sys.spawn([&](event_based_actor* self, actor m) -> behavior { auto helper_actor = sys.spawn([&](event_based_actor* self, actor m) -> behavior {
return { return {
[=](const std::string& str) { [=](const std::string& str) {
CAF_LOG_DEBUG("received '" << str << "'"); std::cerr << "received '" << str << "'" << std::endl;
self->send(m, quit_atom::value); self->send(m, quit_atom::value);
}, },
[=](actor a) { [=](actor a) {
CAF_LOG_DEBUG("got new newb handle"); std::cerr << "got new newb handle" << std::endl;
self->send(m, a); self->send(m, a);
}, },
[=](quit_atom) { [=](quit_atom) {
CAF_LOG_DEBUG("helper shutting down"); std::cerr << "helper shutting down" << std::endl;
self->quit(); self->quit();
} }
}; };
}, main_actor); }, main_actor);
CAF_LOG_DEBUG("creating new acceptor"); std::cerr << "creating new acceptor" << std::endl;
auto newb_acceptor_ptr auto newb_acceptor_ptr
= io::network::make_newb_acceptor<udp_newb_acceptor_t, = io::network::make_server_newb<udp_newb_acceptor_t,
udp_accept_policy_t>(sys, port); udp_accept_policy_t>(sys, port);
dynamic_cast<udp_newb_acceptor_t*>(newb_acceptor_ptr.get())->responder dynamic_cast<udp_newb_acceptor_t*>(newb_acceptor_ptr.get())->responder
= helper_actor; = helper_actor;
CAF_LOG_DEBUG("contacting from 'old-style' broker"); std::cerr << "contacting from 'old-style' broker" << std::endl;
auto test_broker = sys.middleman().spawn_broker(testing, host, port, main_actor); auto test_broker = sys.middleman().spawn_broker(testing, host, port, main_actor);
main_actor->send(test_broker, send_atom::value, "hello world"); main_actor->send(test_broker, send_atom::value, "hello world");
std::this_thread::sleep_for(std::chrono::seconds(1)); std::this_thread::sleep_for(std::chrono::seconds(1));
...@@ -723,27 +631,27 @@ void caf_main(actor_system& sys, const actor_system_config&) { ...@@ -723,27 +631,27 @@ void caf_main(actor_system& sys, const actor_system_config&) {
newb_actor = a; newb_actor = a;
} }
); );
CAF_LOG_DEBUG("new newb was created"); std::cerr << "new newb was created" << std::endl;
main_actor->receive( main_actor->receive(
[](quit_atom) { [](quit_atom) {
CAF_LOG_DEBUG("check"); std::cerr << "check" << std::endl;
} }
); );
CAF_LOG_DEBUG("sending message from newb"); std::cerr << "sending message from newb" << std::endl;
main_actor->send(newb_actor, send_atom::value, actor_id{3}, actor_id{4}, main_actor->send(newb_actor, send_atom::value, actor_id{3}, actor_id{4},
"dlrow olleh"); "dlrow olleh");
main_actor->receive( main_actor->receive(
[](quit_atom) { [](quit_atom) {
CAF_LOG_DEBUG("check"); std::cerr << "check" << std::endl;
} }
); );
CAF_LOG_DEBUG("shutting everything down"); std::cerr << "shutting everything down" << std::endl;
newb_acceptor_ptr->stop(); newb_acceptor_ptr->stop();
anon_send(newb_actor, quit_atom::value); anon_send(newb_actor, quit_atom::value);
anon_send(helper_actor, quit_atom::value); anon_send(helper_actor, quit_atom::value);
anon_send(test_broker, quit_atom::value); anon_send(test_broker, quit_atom::value);
sys.await_all_actors_done(); sys.await_all_actors_done();
CAF_LOG_DEBUG("done"); std::cerr << "done" << std::endl;
} }
} // namespace anonymous } // namespace anonymous
......
...@@ -136,6 +136,12 @@ struct transport_policy { ...@@ -136,6 +136,12 @@ struct transport_policy {
return none; return none;
} }
virtual expected<native_socket>
connect(const std::string&, uint16_t,
optional<io::network::protocol::network> = none) {
return sec::bad_function_call;
}
size_t received_bytes; size_t received_bytes;
size_t max_consecutive_reads; size_t max_consecutive_reads;
...@@ -154,6 +160,9 @@ struct accept_policy { ...@@ -154,6 +160,9 @@ struct accept_policy {
// nop // nop
} }
virtual expected<native_socket> create_socket(uint16_t port, const char* host,
bool reuse = false) = 0;
virtual std::pair<native_socket, transport_policy_ptr> virtual std::pair<native_socket, transport_policy_ptr>
accept(network::event_handler*) = 0; accept(network::event_handler*) = 0;
...@@ -223,14 +232,15 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -223,14 +232,15 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
newb(actor_config& cfg, default_multiplexer& dm, native_socket sockfd) newb(actor_config& cfg, default_multiplexer& dm, native_socket sockfd)
: super(cfg), : super(cfg),
event_handler(dm, sockfd) { event_handler(dm, sockfd) {
// nop CAF_LOG_TRACE("");
} }
newb() = default; newb() = default;
newb(newb<Message>&&) = default; newb(newb<Message>&&) = default;
~newb() override { ~newb() override {
// nop CAF_LOG_TRACE("");
} }
// -- overridden modifiers of abstract_actor --------------------------------- // -- overridden modifiers of abstract_actor ---------------------------------
...@@ -298,6 +308,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -298,6 +308,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
// -- overridden modifiers of event handler ---------------------------------- // -- overridden modifiers of event handler ----------------------------------
void handle_event(network::operation op) override { void handle_event(network::operation op) override {
//std::cerr << "got event: " << to_string(op) << std::endl;
CAF_PUSH_AID_FROM_PTR(this); CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
switch (op) { switch (op) {
...@@ -334,6 +345,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -334,6 +345,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
void start() { void start() {
CAF_PUSH_AID_FROM_PTR(this); CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
intrusive_ptr_add_ref(super::ctrl());
CAF_LOG_DEBUG("starting newb"); CAF_LOG_DEBUG("starting newb");
activate(); activate();
if (transport) if (transport)
...@@ -343,6 +355,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template ...@@ -343,6 +355,7 @@ struct newb : public extend<scheduled_actor, newb<Message>>::template
void stop() { void stop() {
CAF_PUSH_AID_FROM_PTR(this); CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
intrusive_ptr_release(super::ctrl());
close_read_channel(); close_read_channel();
passivate(); passivate();
} }
...@@ -493,8 +506,6 @@ struct newb_acceptor : public network::event_handler { ...@@ -493,8 +506,6 @@ struct newb_acceptor : public network::event_handler {
virtual expected<actor> create_newb(native_socket sock, virtual expected<actor> create_newb(native_socket sock,
transport_policy_ptr pol) = 0; transport_policy_ptr pol) = 0;
// TODO: Has to implement a static create socket function ...
std::unique_ptr<accept_policy<Message>> acceptor; std::unique_ptr<accept_policy<Message>> acceptor;
}; };
...@@ -508,21 +519,37 @@ actor make_newb(actor_system& sys, native_socket sockfd) { ...@@ -508,21 +519,37 @@ actor make_newb(actor_system& sys, native_socket sockfd) {
return actor_cast<actor>(res); return actor_cast<actor>(res);
} }
// TODO: I feel like this should include the ProtocolPolicy somehow. template <class Newb, class Transport, class Protocol>
actor make_client_newb(actor_system& sys, std::string host, uint16_t port) {
transport_policy_ptr trans{new Transport};
expected<native_socket> esock = trans->connect(host, port);
if (!esock)
return {};
auto res = make_newb<Newb>(sys, *esock);
auto ptr = caf::actor_cast<caf::abstract_actor*>(res);
CAF_ASSERT(ptr != nullptr);
auto& ref = dynamic_cast<Newb&>(*ptr);
ref.transport = std::move(trans);
ref.protocol.reset(new Protocol(&ref));
ref.start();
return res;
}
template <class NewbAcceptor, class AcceptPolicy> template <class NewbAcceptor, class AcceptPolicy>
std::unique_ptr<NewbAcceptor> make_newb_acceptor(actor_system& sys, std::unique_ptr<NewbAcceptor> make_server_newb(actor_system& sys,
uint16_t port, uint16_t port,
const char* addr = nullptr, const char* addr = nullptr,
bool reuse_addr = false) { bool reuse_addr = false) {
auto sockfd = NewbAcceptor::create_socket(port, addr, reuse_addr); std::unique_ptr<AcceptPolicy> acc{new AcceptPolicy};
auto esock = acc->create_socket(port, addr, reuse_addr);
// new_tcp_acceptor_impl(port, addr, reuse_addr); // new_tcp_acceptor_impl(port, addr, reuse_addr);
if (!sockfd) { if (!esock) {
CAF_LOG_DEBUG("Could not open " << CAF_ARG(port) << CAF_ARG(addr)); CAF_LOG_DEBUG("Could not open " << CAF_ARG(port) << CAF_ARG(addr));
return nullptr; return nullptr;
} }
auto& mpx = dynamic_cast<default_multiplexer&>(sys.middleman().backend()); auto& mpx = dynamic_cast<default_multiplexer&>(sys.middleman().backend());
std::unique_ptr<NewbAcceptor> ptr{new NewbAcceptor(mpx, *sockfd)}; std::unique_ptr<NewbAcceptor> ptr{new NewbAcceptor(mpx, *esock)};
ptr->acceptor.reset(new AcceptPolicy); ptr->acceptor = std::move(acc);
ptr->start(); ptr->start();
return ptr; return ptr;
} }
......
...@@ -303,6 +303,10 @@ struct dummy_basp_newb : network::newb<new_basp_message> { ...@@ -303,6 +303,10 @@ struct dummy_basp_newb : network::newb<new_basp_message> {
}; };
struct accept_policy_impl : public network::accept_policy<new_basp_message> { struct accept_policy_impl : public network::accept_policy<new_basp_message> {
expected<native_socket> create_socket(uint16_t, const char*, bool) override {
return sec::bad_function_call;
}
std::pair<native_socket, network::transport_policy_ptr> std::pair<native_socket, network::transport_policy_ptr>
accept(network::event_handler*) override { accept(network::event_handler*) override {
// TODO: For UDP read the message into a buffer. Create a new socket. // TODO: For UDP read the message into a buffer. Create a new socket.
......
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