changing error to something with more generic codes

driver/io-unix.cpp

@@ -32,11 +32,11 @@ UnixIoStream::UnixIoStream(UnixEventPort &event_port, int file_descriptor,
 }
 
 ErrorOr<size_t> UnixIoStream::read(void *buffer, size_t length) {
-	ssize_t read_bytes =  unixRead(fd(), buffer, length);
+	ssize_t read_bytes = unixRead(fd(), buffer, length);
-	if( read_bytes > 0 ){
+	if (read_bytes > 0) {
 		return static_cast<size_t>(read_bytes);
-	}else if(read_bytes == 0){
+	} else if (read_bytes == 0) {
-		return criticalError("Disconnected", static_cast<int8_t>(Error::Type::Disconnected));
+		return criticalError("Disconnected", Error::Code::Disconnected);
 	}
 
 	return recoverableError("Currently busy");
@@ -56,17 +56,17 @@ Conveyor<void> UnixIoStream::onReadDisconnected() {
 
 ErrorOr<size_t> UnixIoStream::write(const void *buffer, size_t length) {
 	ssize_t write_bytes = unixWrite(fd(), buffer, length);
-	if( write_bytes > 0){
+	if (write_bytes > 0) {
 		return static_cast<size_t>(write_bytes);
 	}
 
 	int error = errno;
 
-	if( error == EAGAIN || error == EWOULDBLOCK ){
+	if (error == EAGAIN || error == EWOULDBLOCK) {
 		return recoverableError("Currently busy");
 	}
 
-	return criticalError("Disconnected", static_cast<int8_t>(Error::Type::Disconnected));
+	return criticalError("Disconnected", Error::Code::Disconnected);
 }
 
 Conveyor<void> UnixIoStream::writeReady() {

source/kelgin/common.h

@@ -15,7 +15,9 @@ namespace gin {
 	classname(const classname &) = delete;                                     \
 	classname &operator=(const classname &) = delete
 
-#define GIN_ASSERT(expression) assert(expression) if (!expression)
+#define GIN_ASSERT(expression)                                                 \
+	assert(expression);                                                        \
+	if (!expression)
 
 template <typename T> using Maybe = std::optional<T>;
 

source/kelgin/error.cpp

@@ -1,13 +1,13 @@
 #include "error.h"
 
 namespace gin {
-Error::Error() : error_{static_cast<Error::Type>(0)} {}
+Error::Error() : error_{static_cast<Error::Code>(0)} {}
 
-Error::Error(const std::string_view &msg, int8_t code)
+Error::Error(const std::string_view &msg, Error::Code code)
-	: error_message{msg}, error_{static_cast<Error::Type>(code)} {}
+	: error_message{msg}, error_{static_cast<Error::Code>(code)} {}
 
-Error::Error(std::string &&msg, int8_t code)
+Error::Error(std::string &&msg, Error::Code code)
-	: error_message{std::move(msg)}, error_{static_cast<Error::Type>(code)} {}
+	: error_message{std::move(msg)}, error_{static_cast<Error::Code>(code)} {}
 
 Error::Error(Error &&error)
 	: error_message{std::move(error.error_message)}, error_{std::move(
@@ -30,11 +30,17 @@ const std::string_view Error::message() const {
 		error_message);
 }
 
-bool Error::failed() const { return static_cast<int8_t>(error_) != 0; }
+bool Error::failed() const {
+	return static_cast<std::underlying_type_t<Error::Code>>(error_) != 0;
+}
 
-bool Error::isCritical() const { return static_cast<int8_t>(error_) < 0; }
+bool Error::isCritical() const {
+	return static_cast<std::underlying_type_t<Error::Code>>(error_) < 0;
+}
 
-bool Error::isRecoverable() const { return static_cast<int8_t>(error_) > 0; }
+bool Error::isRecoverable() const {
+	return static_cast<std::underlying_type_t<Error::Code>>(error_) > 0;
+}
 
 Error Error::copyError() const {
 	Error error;
@@ -48,16 +54,14 @@ Error Error::copyError() const {
 	return error;
 }
 
-Error::Type Error::code() const {
+Error::Code Error::code() const { return static_cast<Error::Code>(error_); }
-	return static_cast
+
+Error criticalError(const std::string_view &generic, Error::Code c) {
+	return Error{generic, c};
 }
 
-Error criticalError(const std::string_view &generic) {
+Error recoverableError(const std::string_view &generic, Error::Code c) {
-	return Error{generic, -1};
+	return Error{generic, c};
-}
-
-Error recoverableError(const std::string_view &generic) {
-	return Error{generic, 1};
 }
 
 Error noError() { return Error{}; }

source/kelgin/error.h

@@ -14,17 +14,21 @@ namespace gin {
  */
 class Error {
 public:
-	enum class Type : int8_t {
+	enum class Code : int16_t {
-		Disconnected = -99
+		GenericCritical = -1,
+		GenericRecoverable = 1,
+		Disconnected = -99,
+		Exhausted = 99
 	};
+
 private:
 	std::variant<std::string_view, std::string> error_message;
-	Type error_;
+	Code error_;
 
 public:
 	Error();
-	Error(const std::string_view &msg, int8_t code);
+	Error(const std::string_view &msg, Error::Code code);
-	Error(std::string &&msg, int8_t code);
+	Error(std::string &&msg, Error::Code code);
 	Error(Error &&error);
 
 	GIN_FORBID_COPY(Error);
@@ -39,11 +43,11 @@ public:
 
 	Error copyError() const;
 
-	Type code() const;
+	Code code() const;
 };
 
 template <typename Formatter>
-Error makeError(const Formatter &formatter, int8_t code,
+Error makeError(const Formatter &formatter, Error::Code code,
 				const std::string_view &generic) {
 	try {
 		std::string error_msg = formatter();
@@ -53,20 +57,23 @@ Error makeError(const Formatter &formatter, int8_t code,
 	}
 }
 
-Error criticalError(const std::string_view &generic);
+Error criticalError(const std::string_view &generic,
+					Error::Code c = Error::Code::GenericCritical);
 
 template <typename Formatter>
-Error criticalError(const Formatter &formatter,
+Error criticalError(const Formatter &formatter, const std::string_view &generic,
-					const std::string_view &generic) {
+					Error::Code c = Error::Code::GenericCritical) {
-	return makeError(formatter, -1, generic);
+	return makeError(formatter, c, generic);
 }
 
-Error recoverableError(const std::string_view &generic);
+Error recoverableError(const std::string_view &generic,
+					   Error::Code c = Error::Code::GenericRecoverable);
 
 template <typename Formatter>
 Error recoverableError(const Formatter &formatter,
-					   const std::string_view &generic) {
+					   const std::string_view &generic,
-	return makeError(formatter, -1, generic);
+					   Error::Code c = Error::Code::GenericRecoverable) {
+	return makeError(formatter, c, generic);
 }
 
 Error noError();

source/kelgin/io.cpp

@@ -1,3 +1,34 @@
 #include "io.h"
 
-namespace gin {} // namespace gin
+namespace gin {
+
+AsyncIoStream::AsyncIoStream(Own<IoStream> str)
+	: stream{std::move(str)},
+	  read_ready{stream->readReady().then([this]() {}).sink()},
+	  write_ready{stream->writeReady()
+					  .then([this]() {
+
+					  })
+					  .sink()},
+	  read_disconnected{stream->onReadDisconnected()
+							.then([this]() {
+
+							})
+							.sink()} {}
+
+void AsyncIoStream::read(void *buffer, size_t min_length, size_t max_length) {}
+
+Conveyor<size_t> AsyncIoStream::readDone() {
+	auto caf = newConveyorAndFeeder<size_t>();
+	read_stepper.read_done = std::move(caf.feeder);
+	return std::move(caf.conveyor);
+}
+
+void AsyncIoStream::write(const void *buffer, size_t length) {}
+
+Conveyor<size_t> AsyncIoStream::writeDone() {
+	auto caf = newConveyorAndFeeder<size_t>();
+	write_stepper.write_done = std::move(caf.feeder);
+	return std::move(caf.conveyor);
+}
+} // namespace gin

source/kelgin/io.h

@@ -2,6 +2,7 @@
 
 #include "async.h"
 #include "common.h"
+#include "io_helpers.h"
 
 #include <string>
 
@@ -40,12 +41,11 @@ public:
 	virtual ~IoStream() = default;
 };
 
-/*
 class AsyncInputStream {
 public:
 	virtual ~AsyncInputStream() = default;
 
-	virtual void read(void* buffer, size_t min_length, size_t max_length) = 0;
+	virtual void read(void *buffer, size_t min_length, size_t max_length) = 0;
 
 	virtual Conveyor<size_t> readDone() = 0;
 };
@@ -54,7 +54,7 @@ class AsyncOutputStream {
 public:
 	virtual ~AsyncOutputStream() = default;
 
-	virtual void write(const void* buffer, size_t length) = 0;
+	virtual void write(const void *buffer, size_t length) = 0;
 
 	virtual Conveyor<size_t> writeDone() = 0;
 };
@@ -62,18 +62,25 @@ public:
 class AsyncIoStream : public AsyncInputStream, public AsyncOutputStream {
 private:
 	Own<IoStream> stream;
+
+	SinkConveyor read_ready;
+	SinkConveyor write_ready;
+	SinkConveyor read_disconnected;
+
+	ReadTaskAndStepHelper read_stepper;
+	WriteTaskAndStepHelper write_stepper;
+
 public:
 	AsyncIoStream(Own<IoStream> str);
 
-	void read(void* buffer, size_t min_length, size_t max_length) override;
+	void read(void *buffer, size_t min_length, size_t max_length) override;
 
 	Conveyor<size_t> readDone() override;
 
-	void write(const void* buffer, size_t length) override;
+	void write(const void *buffer, size_t length) override;
 
 	Conveyor<size_t> writeDone() override;
 };
-*/
 
 class Server {
 public:

source/kelgin/io_helpers.cpp

@@ -2,40 +2,43 @@
 
 #include "io.h"
 
+#include <cassert>
+
 namespace gin {
 void ReadTaskAndStepHelper::readStep(InputStream &reader) {
 	if (read_task.has_value()) {
 		ReadIoTask &task = *read_task;
 
-		ssize_t n = reader.read(task.buffer, task.max_length);
+		ErrorOr<size_t> n_err = reader.read(task.buffer, task.max_length);
-
+		if (n_err.isError()) {
-		if (n <= 0) {
+			const Error &error = n_err.error();
-			if (n == 0) {
+			if (error.isCritical()) {
-				if (on_read_disconnect) {
-					on_read_disconnect->feed();
-				}
-				return;
-			}
-			int error = errno;
-			if (error == EAGAIN || error == EWOULDBLOCK) {
-				return;
-			} else {
 				if (read_done) {
-					read_done->fail(criticalError("Read failed"));
+					read_done->fail(error.copyError());
 				}
 				read_task = std::nullopt;
 			}
-		} else if (static_cast<size_t>(n) >= task.min_length &&
+		} else if (n_err.isValue()) {
-				   static_cast<size_t>(n) <= task.max_length) {
+			size_t n = n_err.value();
-			if (read_done) {
+			if (static_cast<size_t>(n) >= task.min_length &&
-				read_done->feed(static_cast<size_t>(n));
+				static_cast<size_t>(n) <= task.max_length) {
+				if (read_done) {
+					// Accumulated bytes are not pushed
+					read_done->feed(n + task.already_read);
+				}
+				read_task = std::nullopt;
+			} else {
+				task.buffer = reinterpret_cast<void *>(task.buffer) + n;
+				task.min_length -= static_cast<size_t>(n);
+				task.max_length -= static_cast<size_t>(n);
+				task.already_read += n;
 			}
-			size_t max_len = task.max_length;
+
-			read_task = std::nullopt;
 		} else {
-			task.buffer = reinterpret_cast<void *>(task.buffer) + n;
+			if (read_done) {
-			task.min_length -= static_cast<size_t>(n);
+				read_done->fail(criticalError("Read failed"));
-			task.max_length -= static_cast<size_t>(n);
+			}
+			read_task = std::nullopt;
 		}
 	}
 }
@@ -44,27 +47,35 @@ void WriteTaskAndStepHelper::writeStep(OutputStream &writer) {
 	if (write_task.has_value()) {
 		WriteIoTask &task = *write_task;
 
-		ssize_t n = writer.write(task.buffer, task.length);
+		ErrorOr<size_t> n_err = writer.write(task.buffer, task.length);
 
-		if (n < 0) {
+		if (n_err.isValue()) {
-			int error = errno;
+			size_t n = n_err.value();
-			if (error == EAGAIN || error == EWOULDBLOCK) {
+			assert(n <= task.length);
-				return;
+			if (n == task.length) {
-			} else {
 				if (write_done) {
-					write_done->fail(criticalError("Write failed"));
+					write_done->feed(n + task.already_written);
+				}
+				write_task = std::nullopt;
+			} else {
+				task.buffer = reinterpret_cast<const void *>(task.buffer) +
+							  static_cast<size_t>(n);
+				task.length -= n;
+				task.already_written += n;
+			}
+		} else if (n_err.isError()) {
+			const Error &error = n_err.error();
+			if (error.isCritical()) {
+				if (write_done) {
+					write_done->fail(error.copyError());
 				}
 				write_task = std::nullopt;
 			}
-		} else if (static_cast<size_t>(n) == task.length) {
+		} else {
 			if (write_done) {
-				write_done->feed(static_cast<size_t>(task.length));
+				write_done->fail(criticalError("Write failed"));
 			}
 			write_task = std::nullopt;
-		} else {
-			task.buffer = reinterpret_cast<const void *>(task.buffer) +
-						  static_cast<size_t>(n);
-			task.length -= static_cast<size_t>(n);
 		}
 	}
 }

source/kelgin/io_helpers.h

@@ -1,7 +1,7 @@
 #pragma once
 
-#include <kelgin/async.h>
+#include "async.h"
-#include <kelgin/common.h>
+#include "common.h"
 
 #include <cstdint>
 #include <optional>
@@ -24,6 +24,7 @@ public:
 		void *buffer;
 		size_t min_length;
 		size_t max_length;
+		size_t already_read = 0;
 	};
 	std::optional<ReadIoTask> read_task;
 	Own<ConveyorFeeder<size_t>> read_done = nullptr;
@@ -41,6 +42,7 @@ public:
 	struct WriteIoTask {
 		const void *buffer;
 		size_t length;
+		size_t already_written = 0;
 	};
 	std::optional<WriteIoTask> write_task;
 	Own<ConveyorFeeder<size_t>> write_done = nullptr;