wip

1 files changed, 20 insertions, 10 deletions

diff --git a/modules/io/c++/io_unix.cpp b/modules/io/c++/io_unix.cpp
index 5a8da3a..2c6cf00 100644
--- a/modules/io/c++/io_unix.cpp
+++ b/modules/io/c++/io_unix.cpp
@@ -738,6 +738,10 @@ own<server> unix_network::listen(network_address &addr) {
 	return heap<unix_server>(event_port_, fd, 0);
 }
 
+}}
+#include <iostream>
+namespace saw { namespace unix {
+
 conveyor<own<io_stream>> unix_network::connect(network_address &addr) {
 	auto unix_addr_storage =
 		translate_network_address_to_unix_network_address(addr);
@@ -762,6 +766,7 @@ conveyor<own<io_stream>> unix_network::connect(network_address &addr) {
 		socket_address &addr_iter = address.unix_address(i);
 		int status =
 			::connect(fd, addr_iter.get_raw(), addr_iter.get_raw_length());
+				
 		if (status < 0) {
 			int error = errno;
 			/*
@@ -772,16 +777,21 @@ conveyor<own<io_stream>> unix_network::connect(network_address &addr) {
 
 			/// @todo Add limit node when implemented
 			if (error == EINPROGRESS) {
-				/*
-				Conveyor<void> write_ready = io_stream->writeReady();
-				return write_ready.then(
-					[ios{std::move(io_stream)}]() mutable {
-						ios->write_ready = nullptr;
-						return std::move(ios);
-					});
-				*/
-				success = true;
-				break;
+				conveyor<void> write_rdy = io_str->write_ready();
+
+				return write_rdy.then([ios = std::move(io_str)] () mutable -> error_or<own<io_stream>> {
+					
+					if(!ios){
+						return make_error<err::invalid_state>("Limit node invalidated");
+					}
+					own<io_stream> mov_ios = std::move(ios);
+					/**
+					 * This guarantees the old async pipe to not be used anymore
+					 */
+					mov_ios->write_ready();
+					
+					return std::move(mov_ios);
+				});
 			} else if (error != EINTR) {
 				/// @todo Push error message from
 				return conveyor<own<io_stream>>{make_error<err::disconnected>()};