summaryrefslogtreecommitdiff
path: root/modules/io/c++/io.cpp
diff options
context:
space:
mode:
authorClaudius "keldu" Holeksa <mail@keldu.de>2024-03-07 10:28:04 +0100
committerClaudius "keldu" Holeksa <mail@keldu.de>2024-03-07 10:28:04 +0100
commit7a097bd3bdb288342cc7314f6942347274811030 (patch)
tree4655ba3f79e79b74ae83ad724d66a56a5e319a1b /modules/io/c++/io.cpp
parentd2520819150d9a794baf7505d7d02dadeacd5266 (diff)
io: Move to new dir structure
Diffstat (limited to 'modules/io/c++/io.cpp')
-rw-r--r--modules/io/c++/io.cpp70
1 files changed, 70 insertions, 0 deletions
diff --git a/modules/io/c++/io.cpp b/modules/io/c++/io.cpp
new file mode 100644
index 0000000..06c9cbb
--- /dev/null
+++ b/modules/io/c++/io.cpp
@@ -0,0 +1,70 @@
+#include "io.hpp"
+
+#include <cassert>
+
+namespace saw {
+
+async_io_stream::async_io_stream(own<io_stream> str)
+ : stream_{std::move(str)},
+ read_ready_{stream_->read_ready()
+ .then([this]() { read_stepper_.read_step(*stream_); })
+ .sink()},
+ write_ready_{stream_->write_ready()
+ .then([this]() { write_stepper_.write_step(*stream_); })
+ .sink()},
+ read_disconnected_{stream_->on_read_disconnected()
+ .then([this]() {
+ if (read_stepper_.on_read_disconnect) {
+ read_stepper_.on_read_disconnect->feed();
+ }
+ })
+ .sink()} {}
+
+void async_io_stream::read(void *buffer, size_t min_length, size_t max_length) {
+ SAW_ASSERT(buffer && max_length >= min_length && min_length > 0) { return; }
+
+ SAW_ASSERT(!read_stepper_.read_task.has_value()) { return; }
+
+ read_stepper_.read_task = read_task_and_step_helper::read_io_task{
+ buffer, min_length, max_length, 0};
+ read_stepper_.read_step(*stream_);
+}
+
+conveyor<size_t> async_io_stream::read_done() {
+ auto caf = new_conveyor_and_feeder<size_t>();
+ read_stepper_.read_done = std::move(caf.feeder);
+ return std::move(caf.conveyor);
+}
+
+conveyor<void> async_io_stream::on_read_disconnected() {
+ auto caf = new_conveyor_and_feeder<void>();
+ read_stepper_.on_read_disconnect = std::move(caf.feeder);
+ return std::move(caf.conveyor);
+}
+
+void async_io_stream::write(const void *buffer, size_t length) {
+ SAW_ASSERT(buffer && length > 0) { return; }
+
+ SAW_ASSERT(!write_stepper_.write_task.has_value()) { return; }
+
+ write_stepper_.write_task =
+ write_task_and_step_helper::write_io_task{buffer, length, 0};
+ write_stepper_.write_step(*stream_);
+}
+
+conveyor<size_t> async_io_stream::write_done() {
+ auto caf = new_conveyor_and_feeder<size_t>();
+ write_stepper_.write_done = std::move(caf.feeder);
+ return std::move(caf.conveyor);
+}
+
+string_network_address::string_network_address(const std::string &address,
+ uint16_t port)
+ : address_value_{address}, port_value_{port} {}
+
+const std::string &string_network_address::address() const {
+ return address_value_;
+}
+
+uint16_t string_network_address::port() const { return port_value_; }
+} // namespace saw