Commit 2690ac42 authored by Joseph Noir's avatar Joseph Noir

Working on datagram propagation

parent 6b717efd
...@@ -70,7 +70,7 @@ private: ...@@ -70,7 +70,7 @@ private:
} // namespace ios } // namespace ios
} // namespace caf } // namespace caf
namespace std{ namespace std {
template<> template<>
struct hash<caf::io::accept_handle> { struct hash<caf::io::accept_handle> {
......
...@@ -49,11 +49,11 @@ public: ...@@ -49,11 +49,11 @@ public:
using super = handle<datagram_sink_handle, invalid_datagram_sink_handle_t>; using super = handle<datagram_sink_handle, invalid_datagram_sink_handle_t>;
constexpr datagram_sink_handle() { datagram_sink_handle() : port_{0} {
// nop // nop
} }
constexpr datagram_sink_handle(const invalid_datagram_sink_handle_t&) { datagram_sink_handle(const invalid_datagram_sink_handle_t&) : port_{0} {
// nop // nop
} }
...@@ -63,10 +63,28 @@ public: ...@@ -63,10 +63,28 @@ public:
return f(meta::type_name("datagram_sink_handle"), x.id_); return f(meta::type_name("datagram_sink_handle"), x.id_);
} }
const std::string& host() const {
return host_;
}
void set_host(std::string host) {
host_ = move(host);
}
uint16_t port() const {
return port_;
}
void set_port(uint16_t port) {
port_ = port;
}
private: private:
inline datagram_sink_handle(int64_t handle_id) : super(handle_id) { inline datagram_sink_handle(int64_t handle_id) : super(handle_id) {
// nop // nop
} }
std::string host_;
uint16_t port_;
}; };
} // namespace io } // namespace io
......
...@@ -254,6 +254,8 @@ public: ...@@ -254,6 +254,8 @@ public:
friend class io::middleman; // disambiguate reference friend class io::middleman; // disambiguate reference
friend class supervisor; friend class supervisor;
using endpoint = std::pair<std::string,uint16_t>;
struct event { struct event {
native_socket fd; native_socket fd;
int mask; int mask;
...@@ -338,6 +340,13 @@ public: ...@@ -338,6 +340,13 @@ public:
void del(operation op, native_socket fd, event_handler* ptr); void del(operation op, native_socket fd, event_handler* ptr);
/// @cond PRIVATE
// Used by datagram senders and receivers to manage known endpoints
std::map<endpoint, datagram_sink_handle>& endpoints();
/// @endcond
private: private:
// platform-dependent additional initialization code // platform-dependent additional initialization code
void init(); void init();
...@@ -397,6 +406,9 @@ private: ...@@ -397,6 +406,9 @@ private:
multiplexer_poll_shadow_data shadow_; multiplexer_poll_shadow_data shadow_;
std::pair<native_socket, native_socket> pipe_; std::pair<native_socket, native_socket> pipe_;
pipe_reader pipe_reader_; pipe_reader pipe_reader_;
// TODO: is this the right place?
// How to maintain endpoints if they close?
std::map<endpoint, datagram_sink_handle> remote_endpoints_;
}; };
inline connection_handle conn_hdl_from_socket(native_socket fd) { inline connection_handle conn_hdl_from_socket(native_socket fd) {
...@@ -629,6 +641,8 @@ public: ...@@ -629,6 +641,8 @@ public:
void handle_event(operation op) override; void handle_event(operation op) override;
std::pair<std::string,uint16_t> get_sender();
private: private:
void prepare_next_read(); void prepare_next_read();
......
...@@ -204,6 +204,17 @@ inspect(Inspector& f, datagram_source_passivated_msg& x) { ...@@ -204,6 +204,17 @@ inspect(Inspector& f, datagram_source_passivated_msg& x) {
return f(meta::type_name("datagram_source_passivated_msg"), x.handle); return f(meta::type_name("datagram_source_passivated_msg"), x.handle);
} }
/// Signalizes that a {@link broker} datagram endpoint was closed.
struct endpoint_closed_msg {
endpoint_handle handle;
};
/// @relates endpoint_closed_msg
template <class Inspector>
typename Inspector::result_type
inspect(Inspector& f, endpoint_closed_msg& x) {
return f(meta::type_name("endpoint_closed_msg"), x.handle);
}
} // namespace io } // namespace io
} // namespace caf } // namespace caf
......
...@@ -622,6 +622,11 @@ void default_multiplexer::del(operation op, native_socket fd, ...@@ -622,6 +622,11 @@ void default_multiplexer::del(operation op, native_socket fd,
new_event(del_flag, op, fd, ptr); new_event(del_flag, op, fd, ptr);
} }
std::map<default_multiplexer::endpoint, datagram_sink_handle>&
default_multiplexer::endpoints() {
return remote_endpoints_;
}
void default_multiplexer::wr_dispatch_request(resumable* ptr) { void default_multiplexer::wr_dispatch_request(resumable* ptr) {
intptr_t ptrval = reinterpret_cast<intptr_t>(ptr); intptr_t ptrval = reinterpret_cast<intptr_t>(ptr);
// on windows, we actually have sockets, otherwise we have file handles // on windows, we actually have sockets, otherwise we have file handles
...@@ -1611,6 +1616,21 @@ void datagram_receiver::handle_event(operation op) { ...@@ -1611,6 +1616,21 @@ void datagram_receiver::handle_event(operation op) {
} }
if (rb == 0) if (rb == 0)
return; return;
auto& endpoints = backend().endpoints();
auto sender = get_sender();
auto endpoint = endpoints.find(sender);
if (endpoint == endpoints.end()) {
auto new_endpoint = backend().add_datagram_sink(reader_->parent(),
sender.first,
sender.second);
if (!new_endpoint) {
// TODO: error handling
CAF_LOG_DEBUG("Could not create endpoint for new sender.");
return;
}
endpoint = endpoints.emplace(sender, *new_endpoint).first;
}
auto hdl = endpoint->second;
auto res = reader_->consume(&backend(), rd_buf_.data(), rb); auto res = reader_->consume(&backend(), rd_buf_.data(), rb);
packet_size_ = rb; packet_size_ = rb;
prepare_next_read(); prepare_next_read();
...@@ -1633,6 +1653,34 @@ void datagram_receiver::handle_event(operation op) { ...@@ -1633,6 +1653,34 @@ void datagram_receiver::handle_event(operation op) {
} }
} }
std::pair<std::string,uint16_t> datagram_receiver::get_sender() {
char addr[INET6_ADDRSTRLEN];
std::string host;
uint16_t port = 0;
switch(last_sender.ss_family) {
case AF_INET:
port = ntohs(reinterpret_cast<sockaddr_in*>(&last_sender)->sin_port);
inet_ntop(AF_INET,
&reinterpret_cast<sockaddr_in*>(&last_sender)->sin_addr,
addr, INET_ADDRSTRLEN);
host.insert(std::begin(host), std::begin(addr),
std::begin(addr) + INET_ADDRSTRLEN);
break;
case AF_INET6:
port = ntohs(reinterpret_cast<sockaddr_in6*>(&last_sender)->sin6_port);
inet_ntop(AF_INET6,
&reinterpret_cast<sockaddr_in*>(&last_sender)->sin_addr,
addr, INET6_ADDRSTRLEN);
host.insert(std::begin(host), std::begin(addr),
std::begin(addr) + INET6_ADDRSTRLEN);
break;
default:
// nop
break;
}
return std::make_pair(std::move(host), port);
}
void datagram_receiver::prepare_next_read() { void datagram_receiver::prepare_next_read() {
rd_buf_.resize(buf_size_); rd_buf_.resize(buf_size_);
} }
......
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