Commit 1fc1d5e5 authored by Joseph Noir's avatar Joseph Noir

Create scribe for remote_actor (handshake missing)

parent d56c07f5
...@@ -225,6 +225,12 @@ public: ...@@ -225,6 +225,12 @@ public:
/// from the `multiplexer` to this broker. /// from the `multiplexer` to this broker.
expected<void> assign_dgram_scribe(dgram_scribe_handle hdl); expected<void> assign_dgram_scribe(dgram_scribe_handle hdl);
/// Assigns a detached `dgram_scribe` instance identified by `hdl`
/// from the `multiplexer` to this broker and assigns it to handle the
/// remote endpoint `host` and `port`.
expected<void> assign_dgram_scribe(dgram_scribe_handle hdl,
const std::string& host, uint16_t port);
/// Creates and assigns a new `dgram_scribe` from given native socked `fd`. /// Creates and assigns a new `dgram_scribe` from given native socked `fd`.
expected<dgram_scribe_handle> add_dgram_scribe(network::native_socket fd); expected<dgram_scribe_handle> add_dgram_scribe(network::native_socket fd);
......
...@@ -120,10 +120,10 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee { ...@@ -120,10 +120,10 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
node_id id; node_id id;
// ports // ports
uint16_t remote_port; uint16_t remote_port;
uint16_t local_port;
// pending operations to be performed after handshake completed // pending operations to be performed after handshake completed
optional<response_promise> callback; optional<response_promise> callback;
// TODO: ordering and reliability things // TODO: ordering
// TODO: reliability
}; };
void set_context(connection_handle hdl); void set_context(connection_handle hdl);
...@@ -136,9 +136,11 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee { ...@@ -136,9 +136,11 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// keeps context information for all open connections // keeps context information for all open connections
std::unordered_map<connection_handle, connection_context> tcp_ctx; std::unordered_map<connection_handle, connection_context> tcp_ctx;
std::unordered_map<dgram_scribe_handle, endpoint_context> udp_ctx;
// points to the current context for callbacks such as `make_proxy` // points to the current context for callbacks such as `make_proxy`
connection_context* this_context = nullptr; connection_context* this_context = nullptr;
endpoint_context* current_context = nullptr;
// stores handles to spawn servers for other nodes; these servers // stores handles to spawn servers for other nodes; these servers
// are spawned whenever the broker learns a new node ID and try to // are spawned whenever the broker learns a new node ID and try to
......
...@@ -304,9 +304,17 @@ public: ...@@ -304,9 +304,17 @@ public:
expected<void> assign_dgram_scribe(abstract_broker* ptr, expected<void> assign_dgram_scribe(abstract_broker* ptr,
dgram_scribe_handle hdl) override; dgram_scribe_handle hdl) override;
expected<void>
assign_dgram_scribe(abstract_broker* ptr, dgram_scribe_handle hdl,
const std::string& host, uint16_t port) override;
dgram_scribe_handle add_dgram_scribe(abstract_broker* ptr, dgram_scribe_handle add_dgram_scribe(abstract_broker* ptr,
native_socket fd) override; native_socket fd) override;
expected<dgram_scribe_handle>
add_dgram_scribe(abstract_broker* ptr, native_socket fd,
const std::string& host, uint16_t port) override;
expected<dgram_scribe_handle> add_dgram_scribe(abstract_broker* ptr, expected<dgram_scribe_handle> add_dgram_scribe(abstract_broker* ptr,
const std::string& host, const std::string& host,
uint16_t port) override; uint16_t port) override;
...@@ -391,7 +399,8 @@ private: ...@@ -391,7 +399,8 @@ private:
// allows initialization of scribes with an exisiting sockaddr // allows initialization of scribes with an exisiting sockaddr
dgram_scribe_handle add_dgram_scribe(abstract_broker* ptr, native_socket fd, dgram_scribe_handle add_dgram_scribe(abstract_broker* ptr, native_socket fd,
sockaddr_storage* addr, size_t len); optional<const sockaddr_storage&> addr,
size_t len, bool is_tmp_endpoint);
//resumable* rd_dispatch_request(); //resumable* rd_dispatch_request();
native_socket epollfd_; // unused in poll() implementation native_socket epollfd_; // unused in poll() implementation
...@@ -608,9 +617,10 @@ public: ...@@ -608,9 +617,10 @@ public:
void handle_event(operation op) override; void handle_event(operation op) override;
void sender_from_sockaddr(sockaddr_storage& sa, size_t len); void sender_from_sockaddr(const sockaddr_storage& sa, size_t len);
void set_endpoint_addr(sockaddr_storage& sa, size_t len); void set_endpoint_addr(const sockaddr_storage& sa, size_t len,
bool is_tmp_endpoint);
private: private:
void prepare_next_read(); void prepare_next_read();
...@@ -632,6 +642,12 @@ private: ...@@ -632,6 +642,12 @@ private:
buffer_type wr_offline_buf_; buffer_type wr_offline_buf_;
// general state // general state
// signifies whether to adapt the sender of the next received messages
// as a new endpoint
bool waiting_for_remote_endpoint;
struct sockaddr_storage remote_endpoint_addr_;
socklen_t remote_endpoint_addr_len_;
// addr of last sender
struct sockaddr_storage sockaddr_; struct sockaddr_storage sockaddr_;
socklen_t sockaddr_len_; socklen_t sockaddr_len_;
std::string host_; std::string host_;
...@@ -687,9 +703,9 @@ public: ...@@ -687,9 +703,9 @@ public:
void handle_event(operation op) override; void handle_event(operation op) override;
void sender_from_sockaddr(sockaddr_storage& sa, size_t len); void sender_from_sockaddr(const sockaddr_storage& sa, size_t len);
std::pair<sockaddr_storage*,size_t> last_sender(); std::pair<const sockaddr_storage&,size_t> last_sender();
private: private:
void prepare_next_read(); void prepare_next_read();
......
...@@ -71,7 +71,8 @@ public: ...@@ -71,7 +71,8 @@ public:
/// Returns a native IPv4 or IPv6 translation of `host`. /// Returns a native IPv4 or IPv6 translation of `host`.
static optional<std::pair<std::string, protocol>> static optional<std::pair<std::string, protocol>>
native_address(const std::string& host, optional<protocol> preferred = none); native_address(const std::string& host, optional<protocol> preferred = none,
optional<uint16_t> port = none);
}; };
} // namespace network } // namespace network
......
...@@ -109,14 +109,28 @@ public: ...@@ -109,14 +109,28 @@ public:
/// Assigns an unbound datagram sink identified by `hdl` to `ptr`. /// Assigns an unbound datagram sink identified by `hdl` to `ptr`.
/// @warning Do not call from outside the multiplexer's event loop. /// @warning Do not call from outside the multiplexer's event loop.
virtual expected<void> virtual expected<void>
assign_dgram_scribe(abstract_broker* ptr, assign_dgram_scribe(abstract_broker* ptr, dgram_scribe_handle hdl) = 0;
dgram_scribe_handle hdl) = 0;
/// Assigns an unbound datagram sink identified by `hdl` to `ptr`,
/// and assigns it to the remote endpoint `host` and `port`.
/// @warning Do not call from outside the multiplexer's event loop.
virtual expected<void>
assign_dgram_scribe(abstract_broker* ptr, dgram_scribe_handle hdl,
const std::string& host, uint16_t port) = 0;
/// Creates a new datagram sink from a native socket handle. /// Creates a new datagram sink from a native socket handle.
/// @warning Do not call from outside the multiplexer's event loop. /// @warning Do not call from outside the multiplexer's event loop.
virtual dgram_scribe_handle virtual dgram_scribe_handle
add_dgram_scribe(abstract_broker* ptr, native_socket fd) = 0; add_dgram_scribe(abstract_broker* ptr, native_socket fd) = 0;
/// Creates a new datagram sink from a native socket handle.
/// @warning Do not call from outside the multiplexer's event loop.
virtual expected<dgram_scribe_handle>
add_dgram_scribe(abstract_broker* ptr, native_socket fd,
const std::string& host, uint16_t port) = 0;
/// Tries to create a datagram sink to sent messages to `host` on `port` /// Tries to create a datagram sink to sent messages to `host` on `port`
/// and returns a new datagram sink managing the endpoint on success. /// and returns a new datagram sink managing the endpoint on success.
/// @warning Do not call from outside the multiplexer's event loop. /// @warning Do not call from outside the multiplexer's event loop.
......
...@@ -65,9 +65,20 @@ public: ...@@ -65,9 +65,20 @@ public:
expected<void> assign_dgram_scribe(abstract_broker* ptr, expected<void> assign_dgram_scribe(abstract_broker* ptr,
dgram_scribe_handle hdl) override; dgram_scribe_handle hdl) override;
expected<void> assign_dgram_scribe(abstract_broker* ptr,
dgram_scribe_handle hdl,
const std::string& host,
uint16_t port) override;
dgram_scribe_handle add_dgram_scribe(abstract_broker* ptr, dgram_scribe_handle add_dgram_scribe(abstract_broker* ptr,
native_socket fd) override; native_socket fd) override;
expected<dgram_scribe_handle>
add_dgram_scribe(abstract_broker* ptr, native_socket fd,
const std::string& host, uint16_t port) override;
expected<dgram_scribe_handle> add_dgram_scribe(abstract_broker* ptr, expected<dgram_scribe_handle> add_dgram_scribe(abstract_broker* ptr,
const std::string& host, const std::string& host,
uint16_t port) override; uint16_t port) override;
......
...@@ -212,6 +212,13 @@ expected<void> abstract_broker::assign_dgram_scribe(dgram_scribe_handle hdl) { ...@@ -212,6 +212,13 @@ expected<void> abstract_broker::assign_dgram_scribe(dgram_scribe_handle hdl) {
return backend().assign_dgram_scribe(this, hdl); return backend().assign_dgram_scribe(this, hdl);
} }
expected<void> abstract_broker::assign_dgram_scribe(dgram_scribe_handle hdl,
const std::string& host,
uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(hdl));
return backend().assign_dgram_scribe(this, hdl, host, port);
}
expected<dgram_scribe_handle> expected<dgram_scribe_handle>
abstract_broker::add_dgram_scribe(network::native_socket fd) { abstract_broker::add_dgram_scribe(network::native_socket fd) {
CAF_LOG_TRACE(CAF_ARG(fd)); CAF_LOG_TRACE(CAF_ARG(fd));
......
...@@ -666,16 +666,14 @@ behavior basp_broker::make_behavior() { ...@@ -666,16 +666,14 @@ behavior basp_broker::make_behavior() {
rp.deliver(std::move(res.error())); rp.deliver(std::move(res.error()));
} }
}, },
[=](connect_atom, dgram_scribe_handle hdl, uint16_t port) { [=](connect_atom, dgram_scribe_handle hdl,
const std::string& host, uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(hdl.id())); CAF_LOG_TRACE(CAF_ARG(hdl.id()));
static_cast<void>(hdl);
static_cast<void>(port);
/*
auto rp = make_response_promise(); auto rp = make_response_promise();
auto res = assign_dgram_scribe(hdl); auto res = assign_dgram_scribe(hdl, host, port);
if (res) { if (res) {
auto& ctx = state.udp_ctx[hdl]; auto& ctx = state.udp_ctx[hdl];
ctx.sink = hdl; ctx.hdl = hdl;
ctx.remote_port = port; ctx.remote_port = port;
ctx.callback = rp; ctx.callback = rp;
// TODO: Start handshake with server as there is no way for // TODO: Start handshake with server as there is no way for
...@@ -685,7 +683,6 @@ behavior basp_broker::make_behavior() { ...@@ -685,7 +683,6 @@ behavior basp_broker::make_behavior() {
<< CAF_ARG(res)); << CAF_ARG(res));
rp.deliver(std::move(res.error())); rp.deliver(std::move(res.error()));
} }
*/
}, },
[=](delete_atom, const node_id& nid, actor_id aid) { [=](delete_atom, const node_id& nid, actor_id aid) {
CAF_LOG_TRACE(CAF_ARG(nid) << ", " << CAF_ARG(aid)); CAF_LOG_TRACE(CAF_ARG(nid) << ", " << CAF_ARG(aid));
......
...@@ -860,12 +860,31 @@ accept_handle default_multiplexer::add_tcp_doorman(abstract_broker* self, ...@@ -860,12 +860,31 @@ accept_handle default_multiplexer::add_tcp_doorman(abstract_broker* self,
dgram_scribe_handle dgram_scribe_handle
default_multiplexer::add_dgram_scribe(abstract_broker* self, default_multiplexer::add_dgram_scribe(abstract_broker* self,
native_socket fd) { native_socket fd) {
return add_dgram_scribe(self, fd, nullptr, 0); return add_dgram_scribe(self, fd, none, 0, false);
}
expected<dgram_scribe_handle>
default_multiplexer::add_dgram_scribe(abstract_broker* self, native_socket fd,
const std::string& host, uint16_t port) {
auto res = interfaces::native_address(host, none, port);
if (!res) {
CAF_LOG_INFO("no such host");
return make_error(sec::cannot_connect_to_node, "no such host", host, port);
}
sockaddr_storage tmp_addr;
auto* const_addr
= reinterpret_cast<const sockaddr_storage*>(
reinterpret_cast<const addrinfo*>(res->first.c_str())->ai_addr);
size_t sockaddr_len = (const_addr->ss_family == AF_INET) ?
sizeof(sockaddr_in) : sizeof(sockaddr_in6);
tmp_addr = *const_addr;
return add_dgram_scribe(self, fd, tmp_addr, sockaddr_len, true);
} }
dgram_scribe_handle dgram_scribe_handle
default_multiplexer::add_dgram_scribe(abstract_broker* self, native_socket fd, default_multiplexer::add_dgram_scribe(abstract_broker* self, native_socket fd,
sockaddr_storage* addr, size_t len) { optional<const sockaddr_storage&> addr,
size_t len, bool is_tmp_endpoint) {
CAF_LOG_TRACE(CAF_ARG(fd)); CAF_LOG_TRACE(CAF_ARG(fd));
CAF_ASSERT(fd != network::invalid_native_socket); CAF_ASSERT(fd != network::invalid_native_socket);
class impl : public dgram_scribe { class impl : public dgram_scribe {
...@@ -921,8 +940,9 @@ default_multiplexer::add_dgram_scribe(abstract_broker* self, native_socket fd, ...@@ -921,8 +940,9 @@ default_multiplexer::add_dgram_scribe(abstract_broker* self, native_socket fd,
void remove_from_loop() override { void remove_from_loop() override {
communicator_.passivate(); communicator_.passivate();
} }
void set_endpoint_addr(sockaddr_storage* addr, size_t len) { void set_endpoint_addr(const sockaddr_storage& addr, size_t len,
communicator_.set_endpoint_addr(*addr, len); bool is_tmp_endpoint) {
communicator_.set_endpoint_addr(addr, len, is_tmp_endpoint);
} }
private: private:
bool launched_; bool launched_;
...@@ -930,7 +950,7 @@ default_multiplexer::add_dgram_scribe(abstract_broker* self, native_socket fd, ...@@ -930,7 +950,7 @@ default_multiplexer::add_dgram_scribe(abstract_broker* self, native_socket fd,
}; };
auto ptr = make_counted<impl>(self, *this, fd); auto ptr = make_counted<impl>(self, *this, fd);
if (addr) if (addr)
ptr->set_endpoint_addr(addr, len); ptr->set_endpoint_addr(*addr, len, is_tmp_endpoint);
self->add_dgram_scribe(ptr); self->add_dgram_scribe(ptr);
return ptr->hdl(); return ptr->hdl();
} }
...@@ -970,8 +990,8 @@ default_multiplexer::add_dgram_doorman(abstract_broker* self, ...@@ -970,8 +990,8 @@ default_multiplexer::add_dgram_doorman(abstract_broker* self,
} }
auto endpoint_info = acceptor_.last_sender(); auto endpoint_info = acceptor_.last_sender();
auto hdl = dm.add_dgram_scribe(parent(), *fd, auto hdl = dm.add_dgram_scribe(parent(), *fd,
endpoint_info.first, endpoint_info.first, endpoint_info.second,
endpoint_info.second); false);
return dgram_doorman::new_endpoint(&dm, hdl); return dgram_doorman::new_endpoint(&dm, hdl);
/* /*
auto hdl = dm.add_dgram_scribe(parent(), acceptor_.host(), auto hdl = dm.add_dgram_scribe(parent(), acceptor_.host(),
...@@ -1092,6 +1112,20 @@ expected<void> default_multiplexer::assign_dgram_scribe(abstract_broker* self, ...@@ -1092,6 +1112,20 @@ expected<void> default_multiplexer::assign_dgram_scribe(abstract_broker* self,
return unit; return unit;
} }
expected<void>
default_multiplexer::assign_dgram_scribe(abstract_broker* self,
dgram_scribe_handle hdl,
const std::string& host,
uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(self->id()) << CAF_ARG(hdl) << CAF_ARG(host)
<< CAF_ARG(port));
auto res = add_dgram_scribe(self, static_cast<native_socket>(hdl.id()),
host, port);
if (!res)
return std::move(res.error());
return unit;
}
expected<dgram_scribe_handle> expected<dgram_scribe_handle>
default_multiplexer::add_dgram_scribe(abstract_broker* self, default_multiplexer::add_dgram_scribe(abstract_broker* self,
const std::string& host, uint16_t port) { const std::string& host, uint16_t port) {
...@@ -1527,7 +1561,8 @@ dgram_communicator::dgram_communicator(default_multiplexer& backend_ref, ...@@ -1527,7 +1561,8 @@ dgram_communicator::dgram_communicator(default_multiplexer& backend_ref,
: event_handler(backend_ref, sockfd), : event_handler(backend_ref, sockfd),
dgram_size_(0), dgram_size_(0),
ack_writes_(false), ack_writes_(false),
writing_(false) { writing_(false),
waiting_for_remote_endpoint(true) {
// TODO: Set some reasonable default. // TODO: Set some reasonable default.
configure_datagram_size(1500); configure_datagram_size(1500);
} }
...@@ -1596,9 +1631,11 @@ void dgram_communicator::handle_event(operation op) { ...@@ -1596,9 +1631,11 @@ void dgram_communicator::handle_event(operation op) {
passivate(); passivate();
return; return;
} }
if (host_.empty()) { if (waiting_for_remote_endpoint) {
sender_from_sockaddr(sockaddr_, sockaddr_len_); sender_from_sockaddr(sockaddr_, sockaddr_len_);
// TODO: connect to endpoint? remote_endpoint_addr_ = std::move(sockaddr_);
remote_endpoint_addr_len_ = sockaddr_len_;
sockaddr_len_ = 0;
} }
if (rb == 0) if (rb == 0)
return; return;
...@@ -1614,7 +1651,7 @@ void dgram_communicator::handle_event(operation op) { ...@@ -1614,7 +1651,7 @@ void dgram_communicator::handle_event(operation op) {
case operation::write: { case operation::write: {
size_t wb; // written bytes size_t wb; // written bytes
if (!send_datagram(wb, fd(), wr_buf_.data(), wr_buf_.size(), if (!send_datagram(wb, fd(), wr_buf_.data(), wr_buf_.size(),
sockaddr_, sockaddr_len_)) { remote_endpoint_addr_, remote_endpoint_addr_len_)) {
writer_->io_failure(&backend(), operation::write); writer_->io_failure(&backend(), operation::write);
backend().del(operation::write, fd(), this); backend().del(operation::write, fd(), this);
} else if (wb > 0) { } else if (wb > 0) {
...@@ -1656,21 +1693,22 @@ void dgram_communicator::prepare_next_read() { ...@@ -1656,21 +1693,22 @@ void dgram_communicator::prepare_next_read() {
rd_buf_.resize(dgram_size_); rd_buf_.resize(dgram_size_);
} }
void dgram_communicator::sender_from_sockaddr(sockaddr_storage& sa, size_t) { void dgram_communicator::sender_from_sockaddr(const sockaddr_storage& sa,
size_t) {
char addr[INET6_ADDRSTRLEN]; char addr[INET6_ADDRSTRLEN];
switch(sa.ss_family) { switch(sa.ss_family) {
case AF_INET: case AF_INET:
port_ = ntohs(reinterpret_cast<sockaddr_in*>(&sa)->sin_port); port_ = ntohs(reinterpret_cast<const sockaddr_in*>(&sa)->sin_port);
inet_ntop(AF_INET, inet_ntop(AF_INET,
&reinterpret_cast<sockaddr_in*>(&sa)->sin_addr, &reinterpret_cast<const sockaddr_in*>(&sa)->sin_addr,
addr, INET_ADDRSTRLEN); addr, INET_ADDRSTRLEN);
host_.insert(std::begin(host_), std::begin(addr), host_.insert(std::begin(host_), std::begin(addr),
std::begin(addr) + INET_ADDRSTRLEN); std::begin(addr) + INET_ADDRSTRLEN);
break; break;
case AF_INET6: case AF_INET6:
port_ = ntohs(reinterpret_cast<sockaddr_in6*>(&sa)->sin6_port); port_ = ntohs(reinterpret_cast<const sockaddr_in6*>(&sa)->sin6_port);
inet_ntop(AF_INET6, inet_ntop(AF_INET6,
&reinterpret_cast<sockaddr_in*>(&sa)->sin_addr, &reinterpret_cast<const sockaddr_in*>(&sa)->sin_addr,
addr, INET6_ADDRSTRLEN); addr, INET6_ADDRSTRLEN);
host_.insert(std::begin(host_), std::begin(addr), host_.insert(std::begin(host_), std::begin(addr),
std::begin(addr) + INET6_ADDRSTRLEN); std::begin(addr) + INET6_ADDRSTRLEN);
...@@ -1681,9 +1719,12 @@ void dgram_communicator::sender_from_sockaddr(sockaddr_storage& sa, size_t) { ...@@ -1681,9 +1719,12 @@ void dgram_communicator::sender_from_sockaddr(sockaddr_storage& sa, size_t) {
} }
} }
void dgram_communicator::set_endpoint_addr(sockaddr_storage& sa, size_t len) { void dgram_communicator::set_endpoint_addr(const sockaddr_storage& sa,
sockaddr_ = sa; size_t len,
sockaddr_len_ = len; bool is_tmp_endpoint) {
waiting_for_remote_endpoint = !is_tmp_endpoint;
remote_endpoint_addr_ = sa;
remote_endpoint_addr_len_ = len;
} }
dgram_acceptor::dgram_acceptor(default_multiplexer& backend_ref, dgram_acceptor::dgram_acceptor(default_multiplexer& backend_ref,
...@@ -1773,21 +1814,21 @@ void dgram_acceptor::handle_event(operation op) { ...@@ -1773,21 +1814,21 @@ void dgram_acceptor::handle_event(operation op) {
} }
} }
void dgram_acceptor::sender_from_sockaddr(sockaddr_storage& sa, size_t) { void dgram_acceptor::sender_from_sockaddr(const sockaddr_storage& sa, size_t) {
char addr[INET6_ADDRSTRLEN]; char addr[INET6_ADDRSTRLEN];
switch(sa.ss_family) { switch(sa.ss_family) {
case AF_INET: case AF_INET:
port_ = ntohs(reinterpret_cast<sockaddr_in*>(&sa)->sin_port); port_ = ntohs(reinterpret_cast<const sockaddr_in*>(&sa)->sin_port);
inet_ntop(AF_INET, inet_ntop(AF_INET,
&reinterpret_cast<sockaddr_in*>(&sa)->sin_addr, &reinterpret_cast<const sockaddr_in*>(&sa)->sin_addr,
addr, INET_ADDRSTRLEN); addr, INET_ADDRSTRLEN);
host_.insert(std::begin(host_), std::begin(addr), host_.insert(std::begin(host_), std::begin(addr),
std::begin(addr) + INET_ADDRSTRLEN); std::begin(addr) + INET_ADDRSTRLEN);
break; break;
case AF_INET6: case AF_INET6:
port_ = ntohs(reinterpret_cast<sockaddr_in6*>(&sa)->sin6_port); port_ = ntohs(reinterpret_cast<const sockaddr_in6*>(&sa)->sin6_port);
inet_ntop(AF_INET6, inet_ntop(AF_INET6,
&reinterpret_cast<sockaddr_in*>(&sa)->sin_addr, &reinterpret_cast<const sockaddr_in*>(&sa)->sin_addr,
addr, INET6_ADDRSTRLEN); addr, INET6_ADDRSTRLEN);
host_.insert(std::begin(host_), std::begin(addr), host_.insert(std::begin(host_), std::begin(addr),
std::begin(addr) + INET6_ADDRSTRLEN); std::begin(addr) + INET6_ADDRSTRLEN);
...@@ -1798,8 +1839,8 @@ void dgram_acceptor::sender_from_sockaddr(sockaddr_storage& sa, size_t) { ...@@ -1798,8 +1839,8 @@ void dgram_acceptor::sender_from_sockaddr(sockaddr_storage& sa, size_t) {
} }
} }
std::pair<sockaddr_storage*,size_t> dgram_acceptor::last_sender() { std::pair<const sockaddr_storage&,size_t> dgram_acceptor::last_sender() {
return std::make_pair(&sockaddr_, sockaddr_len_); return std::make_pair(std::ref(sockaddr_), sockaddr_len_);
} }
void dgram_acceptor::prepare_next_read() { void dgram_acceptor::prepare_next_read() {
...@@ -2048,8 +2089,12 @@ new_dgram_doorman_impl(uint16_t port, const char* addr, bool reuse_addr) { ...@@ -2048,8 +2089,12 @@ new_dgram_doorman_impl(uint16_t port, const char* addr, bool reuse_addr) {
protocol proto = ipv6; protocol proto = ipv6;
if (addr) { if (addr) {
auto addrs = interfaces::native_address(addr); auto addrs = interfaces::native_address(addr);
if (!addrs) if (!addrs) {
//CAF_LOG_INFO("no such host");
//return make_error(sec::cannot_connect_to_node, "no such host",
// addr, port);
return make_error(sec::cannot_open_port, "Invalid ADDR", addr); return make_error(sec::cannot_open_port, "Invalid ADDR", addr);
}
proto = addrs->second; proto = addrs->second;
CAF_ASSERT(proto == ipv4 || proto == ipv6); CAF_ASSERT(proto == ipv4 || proto == ipv6);
} }
......
...@@ -218,15 +218,22 @@ std::vector<std::string> interfaces::list_addresses(protocol proc, ...@@ -218,15 +218,22 @@ std::vector<std::string> interfaces::list_addresses(protocol proc,
optional<std::pair<std::string, protocol>> optional<std::pair<std::string, protocol>>
interfaces::native_address(const std::string& host, interfaces::native_address(const std::string& host,
optional<protocol> preferred) { optional<protocol> preferred,
optional<uint16_t> port) {
addrinfo hint; addrinfo hint;
memset(&hint, 0, sizeof(hint)); memset(&hint, 0, sizeof(hint));
hint.ai_socktype = SOCK_STREAM; hint.ai_socktype = SOCK_STREAM;
if (preferred) if (preferred)
hint.ai_family = *preferred == protocol::ipv4 ? AF_INET : AF_INET6; hint.ai_family = *preferred == protocol::ipv4 ? AF_INET : AF_INET6;
addrinfo* tmp = nullptr; addrinfo* tmp = nullptr;
if (port) {
auto port_str = std::to_string(*port);
if (getaddrinfo(host.c_str(), port_str.c_str(), &hint, &tmp))
return none;
} else {
if (getaddrinfo(host.c_str(), nullptr, &hint, &tmp)) if (getaddrinfo(host.c_str(), nullptr, &hint, &tmp))
return none; return none;
}
std::unique_ptr<addrinfo, decltype(freeaddrinfo)*> addrs{tmp, freeaddrinfo}; std::unique_ptr<addrinfo, decltype(freeaddrinfo)*> addrs{tmp, freeaddrinfo};
char buffer[INET6_ADDRSTRLEN]; char buffer[INET6_ADDRSTRLEN];
for (auto i = addrs.get(); i != nullptr; i = i->ai_next) { for (auto i = addrs.get(); i != nullptr; i = i->ai_next) {
......
...@@ -173,7 +173,7 @@ public: ...@@ -173,7 +173,7 @@ public:
auto hdl = *y; auto hdl = *y;
std::vector<response_promise> tmp{std::move(rp)}; std::vector<response_promise> tmp{std::move(rp)};
pending_.emplace(u, std::move(tmp)); pending_.emplace(u, std::move(tmp));
request(broker_, infinite, connect_atom::value, hdl, port).then( request(broker_, infinite, connect_atom::value, hdl, host, port).then(
[=](node_id& nid, strong_actor_ptr& addr, mpi_set& sigs) { [=](node_id& nid, strong_actor_ptr& addr, mpi_set& sigs) {
auto i = pending_.find(u); auto i = pending_.find(u);
if (i == pending_.end()) if (i == pending_.end())
......
...@@ -227,6 +227,13 @@ test_multiplexer::new_dgram_scribe(const std::string& host, ...@@ -227,6 +227,13 @@ test_multiplexer::new_dgram_scribe(const std::string& host,
return result; return result;
} }
expected<void> test_multiplexer::assign_dgram_scribe(abstract_broker* ptr,
dgram_scribe_handle hdl,
const std::string&,
uint16_t) {
return assign_dgram_scribe(ptr, hdl);
}
expected<void> test_multiplexer::assign_dgram_scribe(abstract_broker* ptr, expected<void> test_multiplexer::assign_dgram_scribe(abstract_broker* ptr,
dgram_scribe_handle hdl) { dgram_scribe_handle hdl) {
class impl : public dgram_scribe { class impl : public dgram_scribe {
...@@ -297,6 +304,14 @@ dgram_scribe_handle test_multiplexer::add_dgram_scribe(abstract_broker*, ...@@ -297,6 +304,14 @@ dgram_scribe_handle test_multiplexer::add_dgram_scribe(abstract_broker*,
abort(); abort();
} }
expected<dgram_scribe_handle>
test_multiplexer::add_dgram_scribe(abstract_broker*, native_socket,
const std::string&, uint16_t) {
std::cerr << "test_multiplexer::add_dgram_scribe called with native socket"
<< std::endl;
abort();
}
expected<dgram_scribe_handle> expected<dgram_scribe_handle>
test_multiplexer::add_dgram_scribe(abstract_broker* ptr, test_multiplexer::add_dgram_scribe(abstract_broker* ptr,
const std::string& host, uint16_t prt) { const std::string& host, uint16_t prt) {
......
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