async, io: Fixing async tests for io

2 files changed, 30 insertions, 9 deletions

diff --git a/modules/async/c++/async.hpp b/modules/async/c++/async.hpp
index 257fbd7..b56742c 100644
--- a/modules/async/c++/async.hpp
+++ b/modules/async/c++/async.hpp
@@ -166,10 +166,12 @@ public:
 	error operator()(error &&err);
 };
 
+/**
+ * Helper object holding a sink_conveyor_node<void>
+ */
 class conveyor_sink {
 private:
 	own<conveyor_node> node_;
-
 public:
 	conveyor_sink();
 	conveyor_sink(own<conveyor_node> &&node);
@@ -246,7 +248,7 @@ public:
 	[[nodiscard]] conveyor<T> limit(size_t val = 1);
 
 	/**
-	 *
+	 * @todo implement
 	 */
 	[[nodiscard]] std::pair<conveyor<T>, merge_conveyor<T>> merge();
 
@@ -257,9 +259,10 @@ public:
 	 */
 	template <typename ErrorFunc = propagate_error>
 	void detach(ErrorFunc &&err_func = propagate_error());
+
 	/**
 	 * Creates a local sink which drops elements, but lifetime control remains
-	 * in your hand.
+	 * in your hand contrary to detach().
 	 */
 	template <typename ErrorFunc = propagate_error>
 	[[nodiscard]] conveyor_sink
diff --git a/modules/async/c++/async.tmpl.hpp b/modules/async/c++/async.tmpl.hpp
index ba2a0b3..98573b5 100644
--- a/modules/async/c++/async.tmpl.hpp
+++ b/modules/async/c++/async.tmpl.hpp
@@ -106,14 +106,28 @@ std::pair<conveyor<T>, merge_conveyor<T>> conveyor<T>::merge() {
 						  std::move(node_ref));
 }
 
-template <>
+/**
+ * 
+ */
+template <typename T>
 template <typename ErrorFunc>
-conveyor_sink conveyor<void>::sink(ErrorFunc &&error_func) {
-	conveyor_storage *storage = node_->next_storage();
+conveyor_sink conveyor<T>::sink(ErrorFunc &&error_func) {
+	conveyor<void> conv_then = [&,this](){
+		if constexpr (std::is_same_v<T,void> ){
+			return then([](){}, std::move(error_func));
+		}else{
+			return then([](T&&){}, std::move(error_func));
+		}
+	}();
+	auto nas = conveyor<void>::from_conveyor(std::move(conv_then));
+	assert(nas);
+
+	conveyor_storage *storage = nas->next_storage();
+
 	SAW_ASSERT(storage) { return conveyor_sink{}; }
 
 	own<sink_conveyor_node> sink_node =
-		heap<sink_conveyor_node>(std::move(node_));
+		heap<sink_conveyor_node>(std::move(nas));
 	conveyor_storage *storage_ptr =
 		static_cast<conveyor_storage *>(sink_node.get());
 
@@ -287,11 +301,15 @@ template <typename T> void immediate_conveyor_node<T>::parent_has_fired() {
 		arm_next();
 	}
 }
-
+}
+#include <iostream>
+namespace saw {
 template <typename T> void immediate_conveyor_node<T>::fire() {
-
+	std::cout<<"Immediate fire"<<std::endl;
 	if (parent_) {
+		std::cout<<"Immediate parent:  "<<queued()<<" "<<parent_->space()<<std::endl;
 		parent_->child_has_fired();
+		std::cout<<"Immediate parent2: "<<queued()<<" "<<parent_->space()<<std::endl;
 		if (queued() > 0 && parent_->space() > 0) {
 			arm_last();
 		}