Commit 1d2fa463 authored by ufownl's avatar ufownl

Allow accessing underlying network address for connection handles

parent 1f0e3c52
...@@ -159,6 +159,18 @@ public: ...@@ -159,6 +159,18 @@ public:
/// Creates and assigns a new `doorman` from given native socked `fd`. /// Creates and assigns a new `doorman` from given native socked `fd`.
accept_handle add_tcp_doorman(network::native_socket fd); accept_handle add_tcp_doorman(network::native_socket fd);
/// Returns the remote address associated to `hdl`
/// or empty string if `hdl` is invalid.
std::string remote_addr(connection_handle hdl);
/// Returns the remote port associated to `hdl`
/// or `0` if `hdl` is invalid.
uint16_t remote_port(connection_handle hdl);
/// Returns the local address associated to `hdl`
/// or empty string if `hdl` is invalid.
std::string local_addr(accept_handle hdl);
/// Returns the local port associated to `hdl` or `0` if `hdl` is invalid. /// Returns the local port associated to `hdl` or `0` if `hdl` is invalid.
uint16_t local_port(accept_handle hdl); uint16_t local_port(accept_handle hdl);
......
...@@ -35,7 +35,7 @@ namespace io { ...@@ -35,7 +35,7 @@ namespace io {
/// @ingroup Broker /// @ingroup Broker
class doorman : public network::acceptor_manager { class doorman : public network::acceptor_manager {
public: public:
doorman(abstract_broker* parent, accept_handle hdl, uint16_t local_port); doorman(abstract_broker* parent, accept_handle hdl);
~doorman(); ~doorman();
...@@ -48,10 +48,6 @@ public: ...@@ -48,10 +48,6 @@ public:
// needs to be launched explicitly // needs to be launched explicitly
virtual void launch() = 0; virtual void launch() = 0;
uint16_t port() const {
return port_;
}
protected: protected:
void detach_from_parent() override; void detach_from_parent() override;
...@@ -67,7 +63,6 @@ protected: ...@@ -67,7 +63,6 @@ protected:
accept_handle hdl_; accept_handle hdl_;
message accept_msg_; message accept_msg_;
uint16_t port_;
}; };
} // namespace io } // namespace io
......
...@@ -139,6 +139,7 @@ public: ...@@ -139,6 +139,7 @@ public:
/// Returns the IO socket. /// Returns the IO socket.
Socket& socket_handle() { return fd_; } Socket& socket_handle() { return fd_; }
const Socket& socket_handle() const { return fd_; }
/// Initializes this stream, setting the socket handle to `fd`. /// Initializes this stream, setting the socket handle to `fd`.
void init(Socket fd) { fd_ = std::move(fd); } void init(Socket fd) { fd_ = std::move(fd); }
...@@ -311,6 +312,7 @@ public: ...@@ -311,6 +312,7 @@ public:
/// Returns the IO socket. /// Returns the IO socket.
inline SocketAcceptor& socket_handle() { return accept_fd_; } inline SocketAcceptor& socket_handle() { return accept_fd_; }
inline const SocketAcceptor& socket_handle() const { return accept_fd_; }
/// Returns the accepted socket. This member function should /// Returns the accepted socket. This member function should
/// be called only from the `new_connection` callback. /// be called only from the `new_connection` callback.
......
...@@ -145,13 +145,19 @@ connection_handle asio_multiplexer::add_tcp_scribe(abstract_broker* self, ...@@ -145,13 +145,19 @@ connection_handle asio_multiplexer::add_tcp_scribe(abstract_broker* self,
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
stream_.flush(this); stream_.flush(this);
} }
std::string addr() const override {
return
stream_.socket_handle().remote_endpoint().address().to_string();
}
uint16_t port() const override {
return stream_.socket_handle().remote_endpoint().port();
}
void launch() { void launch() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
CAF_ASSERT(! launched_); CAF_ASSERT(! launched_);
launched_ = true; launched_ = true;
stream_.start(this); stream_.start(this);
} }
private: private:
bool launched_; bool launched_;
stream<Socket> stream_; stream<Socket> stream_;
...@@ -214,8 +220,7 @@ asio_multiplexer::add_tcp_doorman(abstract_broker* self, ...@@ -214,8 +220,7 @@ asio_multiplexer::add_tcp_doorman(abstract_broker* self,
public: public:
impl(abstract_broker* ptr, default_socket_acceptor&& s, impl(abstract_broker* ptr, default_socket_acceptor&& s,
network::asio_multiplexer& am) network::asio_multiplexer& am)
: doorman(ptr, network::accept_hdl_from_socket(s), : doorman(ptr, network::accept_hdl_from_socket(s)),
s.local_endpoint().port()),
acceptor_(am, s.get_io_service()) { acceptor_(am, s.get_io_service()) {
acceptor_.init(std::move(s)); acceptor_.init(std::move(s));
} }
...@@ -236,7 +241,13 @@ asio_multiplexer::add_tcp_doorman(abstract_broker* self, ...@@ -236,7 +241,13 @@ asio_multiplexer::add_tcp_doorman(abstract_broker* self,
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
acceptor_.start(this); acceptor_.start(this);
} }
std::string addr() const override {
return
acceptor_.socket_handle().local_endpoint().address().to_string();
}
uint16_t port() const override {
return acceptor_.socket_handle().local_endpoint().port();
}
private: private:
network::acceptor<default_socket_acceptor> acceptor_; network::acceptor<default_socket_acceptor> acceptor_;
}; };
......
...@@ -64,6 +64,12 @@ public: ...@@ -64,6 +64,12 @@ public:
/// Called by the underlying IO device to report failures. /// Called by the underlying IO device to report failures.
virtual void io_failure(operation op) = 0; virtual void io_failure(operation op) = 0;
/// Get the address of the underlying IO device.
virtual std::string addr() const = 0;
/// Get the port of the underlying IO device.
virtual uint16_t port() const = 0;
protected: protected:
/// Creates a message signalizing a disconnect to the parent. /// Creates a message signalizing a disconnect to the parent.
virtual message detach_message() = 0; virtual message detach_message() = 0;
......
...@@ -176,6 +176,21 @@ accept_handle abstract_broker::add_tcp_doorman(network::native_socket fd) { ...@@ -176,6 +176,21 @@ accept_handle abstract_broker::add_tcp_doorman(network::native_socket fd) {
return backend().add_tcp_doorman(this, fd); return backend().add_tcp_doorman(this, fd);
} }
std::string abstract_broker::remote_addr(connection_handle hdl) {
auto i = scribes_.find(hdl);
return i != scribes_.end() ? i->second->addr() : std::string{};
}
uint16_t abstract_broker::remote_port(connection_handle hdl) {
auto i = scribes_.find(hdl);
return i != scribes_.end() ? i->second->port() : 0;
}
std::string abstract_broker::local_addr(accept_handle hdl) {
auto i = doormen_.find(hdl);
return i != doormen_.end() ? i->second->addr() : std::string{};
}
uint16_t abstract_broker::local_port(accept_handle hdl) { uint16_t abstract_broker::local_port(accept_handle hdl) {
auto i = doormen_.find(hdl); auto i = doormen_.find(hdl);
return i != doormen_.end() ? i->second->port() : 0; return i != doormen_.end() ? i->second->port() : 0;
......
...@@ -104,7 +104,10 @@ namespace io { ...@@ -104,7 +104,10 @@ namespace io {
namespace network { namespace network {
// helper function // helper function
uint16_t port_of_fd(native_socket fd); std::string local_addr_of_fd(native_socket fd);
uint16_t local_port_of_fd(native_socket fd);
std::string remote_addr_of_fd(native_socket fd);
uint16_t remote_port_of_fd(native_socket fd);
/****************************************************************************** /******************************************************************************
* platform-dependent implementations * * platform-dependent implementations *
...@@ -776,6 +779,12 @@ connection_handle default_multiplexer::add_tcp_scribe(abstract_broker* self, ...@@ -776,6 +779,12 @@ connection_handle default_multiplexer::add_tcp_scribe(abstract_broker* self,
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
stream_.flush(this); stream_.flush(this);
} }
std::string addr() const override {
return remote_addr_of_fd(stream_.fd());
}
uint16_t port() const override {
return remote_port_of_fd(stream_.fd());
}
void launch() { void launch() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
CAF_ASSERT(! launched_); CAF_ASSERT(! launched_);
...@@ -799,7 +808,7 @@ default_multiplexer::add_tcp_doorman(abstract_broker* self, ...@@ -799,7 +808,7 @@ default_multiplexer::add_tcp_doorman(abstract_broker* self,
class impl : public doorman { class impl : public doorman {
public: public:
impl(abstract_broker* ptr, default_socket_acceptor&& s) impl(abstract_broker* ptr, default_socket_acceptor&& s)
: doorman(ptr, network::accept_hdl_from_socket(s), port_of_fd(s.fd())), : doorman(ptr, network::accept_hdl_from_socket(s)),
acceptor_(s.backend()) { acceptor_(s.backend()) {
acceptor_.init(std::move(s)); acceptor_.init(std::move(s));
} }
...@@ -820,6 +829,12 @@ default_multiplexer::add_tcp_doorman(abstract_broker* self, ...@@ -820,6 +829,12 @@ default_multiplexer::add_tcp_doorman(abstract_broker* self,
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
acceptor_.start(this); acceptor_.start(this);
} }
std::string addr() const override {
return local_addr_of_fd(acceptor_.fd());
}
uint16_t port() const override {
return local_port_of_fd(acceptor_.fd());
}
private: private:
network::acceptor<default_socket_acceptor> acceptor_; network::acceptor<default_socket_acceptor> acceptor_;
}; };
...@@ -1217,10 +1232,58 @@ new_tcp_acceptor(uint16_t port, const char* addr, bool reuse) { ...@@ -1217,10 +1232,58 @@ new_tcp_acceptor(uint16_t port, const char* addr, bool reuse) {
bound_port}; bound_port};
} }
uint16_t port_of_fd(native_socket fd) { std::string local_addr_of_fd(native_socket fd) {
sockaddr_storage st;
socklen_t st_len = sizeof(st);
sockaddr* sa = reinterpret_cast<sockaddr*>(&st);
ccall(cc_zero, "getsockname() failed",
getsockname, fd, sa, &st_len);
char addr[INET6_ADDRSTRLEN] {0};
switch (sa->sa_family) {
case AF_INET:
return inet_ntop(AF_INET, &reinterpret_cast<sockaddr_in*>(sa)->sin_addr,
addr, sizeof(addr));
case AF_INET6:
return inet_ntop(AF_INET6, &reinterpret_cast<sockaddr_in6*>(sa)->sin6_addr,
addr, sizeof(addr));
default:
break;
}
throw std::invalid_argument("invalid protocol family");
}
uint16_t local_port_of_fd(native_socket fd) {
sockaddr_storage st;
socklen_t st_len = sizeof(st);
ccall(cc_zero, "getsockname() failed", getsockname,
fd, reinterpret_cast<sockaddr*>(&st), &st_len);
return ntohs(port_of(reinterpret_cast<sockaddr&>(st)));
}
std::string remote_addr_of_fd(native_socket fd) {
sockaddr_storage st;
socklen_t st_len = sizeof(st);
sockaddr* sa = reinterpret_cast<sockaddr*>(&st);
ccall(cc_zero, "getpeername() failed",
getpeername, fd, sa, &st_len);
char addr[INET6_ADDRSTRLEN] {0};
switch (sa->sa_family) {
case AF_INET:
return inet_ntop(AF_INET, &reinterpret_cast<sockaddr_in*>(sa)->sin_addr,
addr, sizeof(addr));
case AF_INET6:
return inet_ntop(AF_INET6, &reinterpret_cast<sockaddr_in6*>(sa)->sin6_addr,
addr, sizeof(addr));
default:
break;
}
throw std::invalid_argument("invalid protocol family");
}
uint16_t remote_port_of_fd(native_socket fd) {
sockaddr_storage st; sockaddr_storage st;
socklen_t st_len = sizeof(st); socklen_t st_len = sizeof(st);
ccall(cc_zero, "getsocketname() failed", getsockname, ccall(cc_zero, "getpeername() failed", getpeername,
fd, reinterpret_cast<sockaddr*>(&st), &st_len); fd, reinterpret_cast<sockaddr*>(&st), &st_len);
return ntohs(port_of(reinterpret_cast<sockaddr&>(st))); return ntohs(port_of(reinterpret_cast<sockaddr&>(st)));
} }
......
...@@ -26,11 +26,10 @@ ...@@ -26,11 +26,10 @@
namespace caf { namespace caf {
namespace io { namespace io {
doorman::doorman(abstract_broker* ptr, accept_handle acc_hdl, uint16_t p) doorman::doorman(abstract_broker* ptr, accept_handle acc_hdl)
: network::acceptor_manager(ptr), : network::acceptor_manager(ptr),
hdl_(acc_hdl), hdl_(acc_hdl),
accept_msg_(make_message(new_connection_msg{hdl_, connection_handle{}})), accept_msg_(make_message(new_connection_msg{hdl_, connection_handle{}})) {
port_(p) {
// nop // nop
} }
......
...@@ -69,6 +69,14 @@ void test_multiplexer::assign_tcp_scribe(abstract_broker* ptr, ...@@ -69,6 +69,14 @@ void test_multiplexer::assign_tcp_scribe(abstract_broker* ptr,
// nop // nop
} }
std::string addr() const override {
return std::string{};
}
uint16_t port() const override {
return 0;
}
private: private:
test_multiplexer* mpx_; test_multiplexer* mpx_;
}; };
...@@ -109,12 +117,12 @@ void test_multiplexer::assign_tcp_doorman(abstract_broker* ptr, ...@@ -109,12 +117,12 @@ void test_multiplexer::assign_tcp_doorman(abstract_broker* ptr,
class impl : public doorman { class impl : public doorman {
public: public:
impl(abstract_broker* self, accept_handle ah, test_multiplexer* mpx) impl(abstract_broker* self, accept_handle ah, test_multiplexer* mpx)
: doorman(self, ah, mpx->port(ah)), : doorman(self, ah),
mpx_(mpx) { mpx_(mpx) {
// nop // nop
} }
void new_connection() { void new_connection() override {
auto& mm = mpx_->pending_connects(); auto& mm = mpx_->pending_connects();
auto i = mm.find(hdl()); auto i = mm.find(hdl());
if (i != mm.end()) { if (i != mm.end()) {
...@@ -125,15 +133,23 @@ void test_multiplexer::assign_tcp_doorman(abstract_broker* ptr, ...@@ -125,15 +133,23 @@ void test_multiplexer::assign_tcp_doorman(abstract_broker* ptr,
} }
} }
void stop_reading() { void stop_reading() override {
mpx_->stopped_reading(hdl()) = true; mpx_->stopped_reading(hdl()) = true;
detach(false); detach(false);
} }
void launch() { void launch() override {
// nop // nop
} }
std::string addr() const override {
return std::string{};
}
uint16_t port() const override {
return mpx_->port(hdl());
}
private: private:
test_multiplexer* mpx_; test_multiplexer* mpx_;
}; };
......
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