Commit dc70223d authored by Joseph Noir's avatar Joseph Noir

Test multiplexer compiles, logic still broken

parent abbd084f
...@@ -59,6 +59,32 @@ public: ...@@ -59,6 +59,32 @@ public:
add_tcp_doorman(abstract_broker* ptr, uint16_t prt, add_tcp_doorman(abstract_broker* ptr, uint16_t prt,
const char* in, bool reuse_addr) override; const char* in, bool reuse_addr) override;
expected<datagram_sink_handle> new_datagram_sink(const std::string& host,
uint16_t port) override;
expected<void> assign_datagram_sink(abstract_broker* ptr,
datagram_sink_handle hdl) override;
datagram_sink_handle add_datagram_sink(abstract_broker* ptr,
native_socket fd) override;
expected<datagram_sink_handle> add_datagram_sink(abstract_broker* ptr,
const std::string& host,
uint16_t port) override;
expected<std::pair<datagram_source_handle, uint16_t>>
new_datagram_source(uint16_t port, const char* in, bool reuse_addr) override;
expected<void> assign_datagram_source(abstract_broker* ptr,
datagram_source_handle hdl) override;
datagram_source_handle add_datagram_source(abstract_broker* ptr,
native_socket fd) override;
expected<std::pair<datagram_source_handle, uint16_t>>
add_datagram_source(abstract_broker* ptr, uint16_t port,
const char* in, bool reuse_addr) override;
supervisor_ptr make_supervisor() override; supervisor_ptr make_supervisor() override;
void run() override; void run() override;
...@@ -67,6 +93,11 @@ public: ...@@ -67,6 +93,11 @@ public:
void provide_acceptor(uint16_t port, accept_handle hdl); void provide_acceptor(uint16_t port, accept_handle hdl);
void provide_datagram_sink(std::string host, uint16_t port,
datagram_sink_handle hdl);
void provide_datagram_source(uint16_t port, datagram_source_handle hdl);
/// A buffer storing bytes. /// A buffer storing bytes.
using buffer_type = std::vector<char>; using buffer_type = std::vector<char>;
...@@ -77,9 +108,15 @@ public: ...@@ -77,9 +108,15 @@ public:
/// Returns the output buffer of the scribe identified by `hdl`. /// Returns the output buffer of the scribe identified by `hdl`.
buffer_type& output_buffer(connection_handle hdl); buffer_type& output_buffer(connection_handle hdl);
/// Returns the output buffer of the datagram source identified by `hdl`.
buffer_type& output_buffer(datagram_sink_handle hdl);
/// Returns the input buffer of the scribe identified by `hdl`. /// Returns the input buffer of the scribe identified by `hdl`.
buffer_type& input_buffer(connection_handle hdl); buffer_type& input_buffer(connection_handle hdl);
/// Returns the input buffer of the datagram source identified by `hdl`.
buffer_type& input_buffer(datagram_source_handle hdl);
/// Returns the configured read policy of the scribe identified by `hdl`. /// Returns the configured read policy of the scribe identified by `hdl`.
receive_policy::config& read_config(connection_handle hdl); receive_policy::config& read_config(connection_handle hdl);
...@@ -97,6 +134,31 @@ public: ...@@ -97,6 +134,31 @@ public:
uint16_t& port(accept_handle hdl); uint16_t& port(accept_handle hdl);
/// Returns `true` if this handle has been closed
/// for reading, `false` otherwise.
bool& stopped_reading(datagram_sink_handle hdl);
/// Returns whether the scribe identified by `hdl` receives write ACKs.
bool& ack_writes(datagram_sink_handle hdl);
/// Returns `true` if this handle is inactive, otherwise `false`.
bool& passive_mode(datagram_sink_handle hdl);
intrusive_ptr<datagram_sink>& impl_ptr(datagram_sink_handle hdl);
uint16_t& port(datagram_sink_handle hdl);
/// Returns `true` if this handle has been closed
/// for reading, `false` otherwise.
bool& stopped_reading(datagram_source_handle hdl);
/// Returns `true` if this handle is inactive, otherwise `false`.
bool& passive_mode(datagram_source_handle hdl);
intrusive_ptr<datagram_source>& impl_ptr(datagram_source_handle hdl);
uint16_t& port(datagram_source_handle hdl);
/// Returns `true` if this handle has been closed /// Returns `true` if this handle has been closed
/// for reading, `false` otherwise. /// for reading, `false` otherwise.
bool& stopped_reading(accept_handle hdl); bool& stopped_reading(accept_handle hdl);
...@@ -119,6 +181,16 @@ public: ...@@ -119,6 +181,16 @@ public:
bool has_pending_scribe(std::string host, uint16_t port); bool has_pending_scribe(std::string host, uint16_t port);
using pending_datagram_sinks_map = std::map<std::pair<std::string, uint16_t>,
datagram_sink_handle>;
bool has_pending_datagram_sink(std::string host, uint16_t port);
using pending_datagram_sources_map = std::map<uint16_t, datagram_source_handle>;
bool has_pending_datagram_source(uint16_t port);
/// Accepts a pending connect on `hdl`. /// Accepts a pending connect on `hdl`.
bool accept_connection(accept_handle hdl); bool accept_connection(accept_handle hdl);
...@@ -167,14 +239,36 @@ private: ...@@ -167,14 +239,36 @@ private:
intrusive_ptr<doorman> ptr; intrusive_ptr<doorman> ptr;
}; };
struct datagram_sink_data {
uint16_t port;
buffer_type wr_buf;
size_t buffer_size;
bool stopped_reading = false;
bool passive_mode = false;
bool ack_writes = false;
intrusive_ptr<datagram_sink> ptr;
};
struct datagram_source_data {
uint16_t port;
buffer_type rd_buf;
bool stopped_reading = false;
bool passive_mode = false;
intrusive_ptr<datagram_source> ptr;
};
// guards resumables_ and scribes_ // guards resumables_ and scribes_
std::mutex mx_; std::mutex mx_;
std::condition_variable cv_; std::condition_variable cv_;
std::list<resumable_ptr> resumables_; std::list<resumable_ptr> resumables_;
pending_scribes_map scribes_; pending_scribes_map scribes_;
pending_datagram_sinks_map datagram_sinks_;
pending_datagram_sources_map datagram_sources_;
std::unordered_map<uint16_t, accept_handle> doormen_; std::unordered_map<uint16_t, accept_handle> doormen_;
std::unordered_map<connection_handle, scribe_data> scribe_data_; std::unordered_map<connection_handle, scribe_data> scribe_data_;
std::unordered_map<accept_handle, doorman_data> doorman_data_; std::unordered_map<accept_handle, doorman_data> doorman_data_;
std::unordered_map<datagram_sink_handle, datagram_sink_data> datagram_sink_data_;
std::unordered_map<datagram_source_handle, datagram_source_data> datagram_source_data_;
pending_connects_map pending_connects_; pending_connects_map pending_connects_;
}; };
......
...@@ -23,6 +23,8 @@ ...@@ -23,6 +23,8 @@
#include "caf/io/scribe.hpp" #include "caf/io/scribe.hpp"
#include "caf/io/doorman.hpp" #include "caf/io/doorman.hpp"
#include "caf/io/datagram_sink.hpp"
#include "caf/io/datagram_source.hpp"
namespace caf { namespace caf {
namespace io { namespace io {
...@@ -212,6 +214,151 @@ test_multiplexer::add_tcp_doorman(abstract_broker* ptr, uint16_t prt, ...@@ -212,6 +214,151 @@ test_multiplexer::add_tcp_doorman(abstract_broker* ptr, uint16_t prt,
return result; return result;
} }
expected<datagram_sink_handle>
test_multiplexer::new_datagram_sink(const std::string& host,
uint16_t port_hint) {
guard_type guard{mx_};
datagram_sink_handle result;
auto i = datagram_sinks_.find(std::make_pair(host, port_hint));
if (i != datagram_sinks_.end()) {
result = i->second;
datagram_sinks_.erase(i);
}
return result;
}
expected<void> test_multiplexer::assign_datagram_sink(abstract_broker* ptr,
datagram_sink_handle hdl) {
class impl : public datagram_sink {
public:
impl(abstract_broker* self, datagram_sink_handle dsh, test_multiplexer* mpx)
: datagram_sink(self, dsh),
mpx_(mpx) {
// nop
}
void stop_reading() override {
mpx_->stopped_reading(hdl()) = true;
detach(mpx_, false);
}
void ack_writes(bool enable) override {
mpx_->ack_writes(hdl()) = enable;
}
std::vector<char>& wr_buf() override {
return mpx_->output_buffer(hdl());
}
std::string addr() const override {
return "test";
}
uint16_t port() const override {
return mpx_->port(hdl());
}
void launch() {
// nop
}
void add_to_loop() override {
mpx_->passive_mode(hdl()) = false;
}
void remove_from_loop() override {
mpx_->passive_mode(hdl()) = true;
}
private:
test_multiplexer* mpx_;
};
auto dsptr = make_counted<impl>(ptr, hdl, this);
ptr->add_datagram_sink(dsptr);
return unit;
}
datagram_sink_handle test_multiplexer::add_datagram_sink(abstract_broker*,
native_socket) {
std::cerr << "test_multiplexer::add_datagram_sink called with native socket"
<< std::endl;
abort();
}
expected<datagram_sink_handle>
test_multiplexer::add_datagram_sink(abstract_broker* ptr,
const std::string& host, uint16_t prt) {
auto result = new_datagram_sink(host, prt);
if (!result)
return std::move(result.error());
// port(*result) = prt; // TODO: remove this?
assign_datagram_sink(ptr, *result);
return result;
}
expected<std::pair<datagram_source_handle, uint16_t>>
test_multiplexer::new_datagram_source(uint16_t desired_port, const char*, bool) {
guard_type guard{mx_};
datagram_source_handle result;
auto i = datagram_sources_.find(desired_port);
if (i != datagram_sources_.end()) {
result = i->second;
datagram_sources_.erase(i);
}
return std::make_pair(result, desired_port);
}
expected<void> test_multiplexer::assign_datagram_source(abstract_broker* ptr,
datagram_source_handle hdl) {
class impl : public datagram_source {
public:
impl(abstract_broker* self, datagram_source_handle dsh, test_multiplexer* mpx)
: datagram_source(self, dsh),
mpx_(mpx) {
// nop
}
void configure_datagram_size(size_t buf_size) override {
}
void stop_reading() override {
mpx_->stopped_reading(hdl()) = true;
detach(mpx_, false);
}
std::vector<char>& rd_buf() override {
return mpx_->input_buffer(hdl());
}
std::string addr() const override {
return "test";
}
uint16_t port() const override {
return mpx_->port(hdl());
}
void launch() {
// nop
}
void add_to_loop() override {
mpx_->passive_mode(hdl()) = false;
}
void remove_from_loop() override {
mpx_->passive_mode(hdl()) = true;
}
private:
test_multiplexer* mpx_;
};
auto dsptr = make_counted<impl>(ptr, hdl, this);
ptr->add_datagram_source(dsptr);
return unit;
}
datagram_source_handle test_multiplexer::add_datagram_source(abstract_broker*,
native_socket) {
std::cerr << "test_multiplexer::add_datagram_source called with native socket"
<< std::endl;
abort();
}
expected<std::pair<datagram_source_handle, uint16_t>>
test_multiplexer::add_datagram_source(abstract_broker* ptr, uint16_t prt,
const char* in, bool reuse_addr) {
auto result = new_datagram_source(prt, in, reuse_addr);
if (!result)
return std::move(result.error());
port(result->first) = prt;
assign_datagram_source(ptr, result->first);
return result;
}
auto test_multiplexer::make_supervisor() -> supervisor_ptr { auto test_multiplexer::make_supervisor() -> supervisor_ptr {
// not needed // not needed
return nullptr; return nullptr;
...@@ -233,6 +380,19 @@ void test_multiplexer::provide_acceptor(uint16_t desired_port, ...@@ -233,6 +380,19 @@ void test_multiplexer::provide_acceptor(uint16_t desired_port,
doorman_data_[hdl].port = desired_port; doorman_data_[hdl].port = desired_port;
} }
void test_multiplexer::provide_datagram_sink(std::string host,
uint16_t desired_port,
datagram_sink_handle hdl) {
datagram_sinks_.emplace(std::make_pair(std::move(host), desired_port), hdl);
datagram_sink_data_[hdl].port = desired_port;
}
void test_multiplexer::provide_datagram_source(uint16_t desired_port,
datagram_source_handle hdl) {
datagram_sources_.emplace(std::make_pair(desired_port, hdl));
datagram_source_data_[hdl].port = desired_port;
}
/// The external input buffer should be filled by /// The external input buffer should be filled by
/// the test program. /// the test program.
test_multiplexer::buffer_type& test_multiplexer::buffer_type&
...@@ -274,6 +434,54 @@ uint16_t& test_multiplexer::port(accept_handle hdl) { ...@@ -274,6 +434,54 @@ uint16_t& test_multiplexer::port(accept_handle hdl) {
return doorman_data_[hdl].port; return doorman_data_[hdl].port;
} }
test_multiplexer::buffer_type&
test_multiplexer::output_buffer(datagram_sink_handle hdl) {
return datagram_sink_data_[hdl].wr_buf;
}
bool& test_multiplexer::stopped_reading(datagram_sink_handle hdl) {
return datagram_sink_data_[hdl].stopped_reading;
}
bool& test_multiplexer::ack_writes(datagram_sink_handle hdl) {
return datagram_sink_data_[hdl].ack_writes;
}
bool& test_multiplexer::passive_mode(datagram_sink_handle hdl) {
return datagram_sink_data_[hdl].passive_mode;
}
intrusive_ptr<datagram_sink>&
test_multiplexer::impl_ptr(datagram_sink_handle hdl) {
return datagram_sink_data_[hdl].ptr;
}
uint16_t& test_multiplexer::port(datagram_sink_handle hdl) {
return datagram_sink_data_[hdl].port;
}
test_multiplexer::buffer_type&
test_multiplexer::input_buffer(datagram_source_handle hdl) {
return datagram_source_data_[hdl].rd_buf;
}
bool& test_multiplexer::stopped_reading(datagram_source_handle hdl) {
return datagram_source_data_[hdl].stopped_reading;
}
bool& test_multiplexer::passive_mode(datagram_source_handle hdl) {
return datagram_source_data_[hdl].passive_mode;
}
intrusive_ptr<datagram_source>&
test_multiplexer::impl_ptr(datagram_source_handle hdl) {
return datagram_source_data_[hdl].ptr;
}
uint16_t& test_multiplexer::port(datagram_source_handle hdl) {
return datagram_source_data_[hdl].port;
}
bool& test_multiplexer::stopped_reading(accept_handle hdl) { bool& test_multiplexer::stopped_reading(accept_handle hdl) {
return doorman_data_[hdl].stopped_reading; return doorman_data_[hdl].stopped_reading;
} }
......
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