From 17431a0c95558ed61f092fa019231df89677ca0f Mon Sep 17 00:00:00 2001
From: Claudius 'keldu' Holeksa <mail@keldu.de>
Date: Fri, 9 Aug 2024 18:26:48 +0200
Subject: wip

---
 modules/io_codec/c++/io_peer.tmpl.hpp | 66 +++++++++--------------------------
 1 file changed, 16 insertions(+), 50 deletions(-)

(limited to 'modules/io_codec/c++/io_peer.tmpl.hpp')

diff --git a/modules/io_codec/c++/io_peer.tmpl.hpp b/modules/io_codec/c++/io_peer.tmpl.hpp
index 967a8ab..0322631 100644
--- a/modules/io_codec/c++/io_peer.tmpl.hpp
+++ b/modules/io_codec/c++/io_peer.tmpl.hpp
@@ -29,29 +29,8 @@ streaming_io_peer<Incoming, Outgoing, TransportEncoding, ContentEncoding,
 		  io_stream_->read_done()
 			  .then([this](size_t bytes) -> error_or<void> {
 
-					std::cout<<"Read done start: "<<bytes<<std::endl;
 				  in_buffer_.write_advance(bytes);
 
-					std::cout<<"buff state: ";
-					std::cout<<in_buffer_.read_composite_length();
-					std::cout<<" ";
-					std::cout<<in_buffer_.read_segment_length();
-					std::cout<<" ";
-					std::cout<<in_buffer_.write_composite_length();
-					std::cout<<" ";
-					std::cout<<in_buffer_.write_segment_length();
-					std::cout<<std::endl;
-					
-					std::cout<<"buff state: ";
-					std::cout<<out_buffer_.read_composite_length();
-					std::cout<<" ";
-					std::cout<<out_buffer_.read_segment_length();
-					std::cout<<" ";
-					std::cout<<out_buffer_.write_composite_length();
-					std::cout<<" ";
-					std::cout<<out_buffer_.write_segment_length();
-					std::cout<<std::endl;
-
 				  if (in_buffer_.write_segment_length() == 0) {
 					  return make_error<err::critical>("Message too long");
 				  }
@@ -82,26 +61,6 @@ streaming_io_peer<Incoming, Outgoing, TransportEncoding, ContentEncoding,
 						in_buffer_.read_advance(len_val);
 						in_buff->write_advance(len_val);
 					
-						std::cout<<"read buff state: ";
-						std::cout<<in_buffer_.read_composite_length();
-						std::cout<<" ";
-						std::cout<<in_buffer_.read_segment_length();
-						std::cout<<" ";
-						std::cout<<in_buffer_.write_composite_length();
-						std::cout<<" ";
-						std::cout<<in_buffer_.write_segment_length();
-						std::cout<<std::endl;
-						
-						std::cout<<"write buff state: ";
-						std::cout<<out_buffer_.read_composite_length();
-						std::cout<<" ";
-						std::cout<<out_buffer_.read_segment_length();
-						std::cout<<" ";
-						std::cout<<out_buffer_.write_composite_length();
-						std::cout<<" ";
-						std::cout<<out_buffer_.write_segment_length();
-						std::cout<<std::endl;
-
 						data<Incoming, ContentEncoding> in_data{std::move(in_buff)};
 						incoming_feeder_->feed(std::move(in_data));
 				  }
@@ -123,7 +82,12 @@ streaming_io_peer<Incoming, Outgoing, TransportEncoding, ContentEncoding,
 						  }
 
 						  return void_t{};
-					  })
+					  },[this](error err){
+							if(disconnect_feeder_){
+								disconnect_feeder_->feed();
+							}
+							return err;
+						})
 					  .sink()
 		},
 		conveyor_feeder_{
@@ -139,10 +103,8 @@ template <typename Incoming, typename Outgoing, typename TransportEncoding,
 error_or<void> streaming_io_peer<Incoming, Outgoing, TransportEncoding, ContentEncoding,
 						BufferT>::send(data<Outgoing, ContentEncoding>
 										   msg) {
-	std::cout<<"O: "<<out_buffer_.read_segment_length()<<std::endl;
 	bool restart_write = (out_buffer_.read_segment_length() == 0u);
 
-	std::cout<<"A: "<<out_buffer_.read_segment_length()<<std::endl;
 
 	auto& msg_buff = msg.get_buffer();
 	{
@@ -153,7 +115,6 @@ error_or<void> streaming_io_peer<Incoming, Outgoing, TransportEncoding, ContentE
 		}
 		// auto& len_val = eov.get_value();
 	}
-	std::cout<<"B"<<std::endl;
 
 	auto eov = out_buffer_.write_from(msg_buff);
 	if (eov.is_error()) {
@@ -162,14 +123,11 @@ error_or<void> streaming_io_peer<Incoming, Outgoing, TransportEncoding, ContentE
 	}
 	auto& len_val = eov.get_value();
 	out_buffer_.write_advance(len_val);
-	std::cout<<"C"<<std::endl;
 
 	if (restart_write && out_buffer_.read_segment_length() > 0u) {
-		std::cout<<"D"<<std::endl;
 		io_stream_->write(&out_buffer_.read(),
 						  out_buffer_.read_segment_length());
 	}
-	std::cout<<"E"<<std::endl;
 
 	return void_t{};
 }
@@ -178,8 +136,16 @@ template <typename Incoming, typename Outgoing, typename TransportEncoding, type
 		  typename BufferT>
 conveyor<void>
 streaming_io_peer<Incoming, Outgoing, TransportEncoding, ContentEncoding,
-				  BufferT>::on_read_disconnected() {
-	return io_stream_->on_read_disconnected();
+				  BufferT>::on_disconnected() {
+	io_read_disconnected_ = io_stream_->on_read_disconnected().then([this](){
+		if(disconnect_feeder_){
+			disconnect_feeder_->feed();
+		}
+	}).sink();
+
+	auto caf = new_conveyor_and_feeder<void>();
+	disconnect_feeder_ = std::move(caf.feeder);
+	return std::move(caf.conveyor);
 }
 
 template <typename Incoming, typename Outgoing, typename TransportEncoding, typename ContentEncoding, typename BufferT>
-- 
cgit v1.2.3