Reworked abstraction for network

1 files changed, 79 insertions, 11 deletions

diff --git a/modules/io/c++/io.hpp b/modules/io/c++/io.hpp
index 390ddcc..fccba7f 100644
--- a/modules/io/c++/io.hpp
+++ b/modules/io/c++/io.hpp
@@ -55,6 +55,7 @@ public:
 /*
  * Io stream
  */
+template<typename T = net::Os>
 class io_stream : public input_stream, public output_stream {
 public:
 	virtual ~io_stream() = default;
@@ -79,10 +80,11 @@ public:
 	virtual conveyor<size_t> write_done() = 0;
 };
 
+template<typename T = net::Os>
 class async_io_stream final : public async_input_stream,
 							  public async_output_stream {
 private:
-	own<io_stream> stream_;
+	own<io_stream<T>> stream_;
 
 	conveyor_sink read_ready_;
 	conveyor_sink write_ready_;
@@ -92,7 +94,7 @@ private:
 	write_task_and_step_helper write_stepper_;
 
 public:
-	async_io_stream(own<io_stream> str);
+	async_io_stream(own<io_stream<T>> str);
 
 	SAW_FORBID_COPY(async_io_stream);
 	SAW_FORBID_MOVE(async_io_stream);
@@ -108,19 +110,22 @@ public:
 	conveyor<size_t> write_done() override;
 };
 
+template<typename T = net::Os>
 class server {
 public:
 	virtual ~server() = default;
 
-	virtual conveyor<own<io_stream>> accept() = 0;
+	virtual conveyor<own<io_stream<T>>> accept() = 0;
 };
 
+template<typename T = net::Os>
 class network_address;
 /**
  * Datagram class. Bound to a local address it is able to receive inbound
  * datagram messages and send them as well as long as an address is provided as
  * well
  */
+template<typename T = net::Os>
 class datagram {
 public:
 	virtual ~datagram() = default;
@@ -129,13 +134,14 @@ public:
 	virtual conveyor<void> read_ready() = 0;
 
 	virtual error_or<size_t> write(const void *buffer, size_t length,
-								   network_address &dest) = 0;
+								   network_address<T> &dest) = 0;
 	virtual conveyor<void> write_ready() = 0;
 };
 
 class os_network_address;
 class string_network_address;
 
+template<typename T>
 class network_address {
 public:
 	using child_variant =
@@ -149,14 +155,14 @@ public:
 	virtual uint16_t port() const = 0;
 };
 
-class os_network_address : public network_address {
+class os_network_address : public network_address<net::Os> {
 public:
 	virtual ~os_network_address() = default;
 
 	network_address::child_variant representation() override { return this; }
 };
 
-class string_network_address final : public network_address {
+class string_network_address final : public network_address<net::Os> {
 private:
 	std::string address_value_;
 	uint16_t port_value_;
@@ -178,10 +184,10 @@ public:
 	/**
 	 * Resolve the provided string and uint16 to the preferred storage method
 	 */
-	virtual conveyor<own<network_address>>
+	virtual conveyor<own<network_address<T>>>
 	resolve_address(const std::string &addr, uint16_t port_hint = 0) = 0;
 	
-	virtual error_or<own<network_address>>
+	virtual error_or<own<network_address<T>>>
 	parse_address(const std::string &addr, uint16_t port_hint = 0) = 0;
 
 	/**
@@ -195,17 +201,17 @@ public:
 	/**
 	 * Set up a listener on this address
 	 */
-	virtual own<server> listen(network_address &bind_addr) = 0;
+	virtual own<server<T>> listen(network_address<T> &bind_addr) = 0;
 
 	/**
 	 * Connect to a remote address
 	 */
-	virtual conveyor<own<io_stream>> connect(network_address &address) = 0;
+	virtual conveyor<own<io_stream<T>>> connect(network_address<T> &address) = 0;
 
 	/**
 	 * Bind a datagram socket at this address.
 	 */
-	virtual own<datagram> datagram(network_address &address) = 0;
+	virtual own<datagram<T>> bind_datagram(network_address<T> &address) = 0;
 };
 
 class io_provider {
@@ -222,6 +228,68 @@ struct async_io_context {
 	event_loop &event_loop;
 	event_port &event_port;
 };
+}
+
+namespace saw {
+template<typename T>
+async_io_stream<T>::async_io_stream(own<io_stream<T>> 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()} {}
+
+template<typename T>
+void async_io_stream<T>::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_);
+}
+
+template<typename T>
+conveyor<size_t> async_io_stream<T>::read_done() {
+	auto caf = new_conveyor_and_feeder<size_t>();
+	read_stepper_.read_done = std::move(caf.feeder);
+	return std::move(caf.conveyor);
+}
+
+template<typename T>
+conveyor<void> async_io_stream<T>::on_read_disconnected() {
+	auto caf = new_conveyor_and_feeder<void>();
+	read_stepper_.on_read_disconnect = std::move(caf.feeder);
+	return std::move(caf.conveyor);
+}
+
+template<typename T>
+void async_io_stream<T>::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_);
+}
+
+template<typename T>
+conveyor<size_t> async_io_stream<T>::write_done() {
+	auto caf = new_conveyor_and_feeder<size_t>();
+	write_stepper_.write_done = std::move(caf.feeder);
+	return std::move(caf.conveyor);
+}
 
 error_or<async_io_context> setup_async_io();
 } // namespace saw